Я хотел иметь возможность телепортироваться или использовать команды и т. п. в игре.
Я обнаружил, что при использовании канала ролевой игры чат регистрируется.
Я нашел способ читать журнал и реагировать на него.
В настоящее время я настроил его так, что вы можете выполнить несколько команд:
rescue = Отправляет предмет для телепортации в Велию
tp velia = Отправляет предмет для телепортации в Велию
tp, а затем: heidel, grana, altinova, aakman, Duvencrune, crescent, harsh, hystria, pirate, sycraia, valencia
Вам по почте будет отправлен предмет, который телепортирует вас в эти места. Это довольно удобно.
1. Вам нужно будет установить Python. Я использую 3.9.5, потому что он установлен с Corsair.
2. Вам понадобится sendmail (на этом же форуме у меня есть пост о Central/Market, Black Market). Под кодом Python описано, как настроить sendMail.
Технически, ограничений нет. Вы можете ввести команду в чат ролевой игры, и эта программа отправит вам все, что угодно).
Я обнаружил, что при использовании канала ролевой игры чат регистрируется.
Я нашел способ читать журнал и реагировать на него.
В настоящее время я настроил его так, что вы можете выполнить несколько команд:
rescue = Отправляет предмет для телепортации в Велию
tp velia = Отправляет предмет для телепортации в Велию
tp, а затем: heidel, grana, altinova, aakman, Duvencrune, crescent, harsh, hystria, pirate, sycraia, valencia
Вам по почте будет отправлен предмет, который телепортирует вас в эти места. Это довольно удобно.
1. Вам нужно будет установить Python. Я использую 3.9.5, потому что он установлен с Corsair.
2. Вам понадобится sendmail (на этом же форуме у меня есть пост о Central/Market, Black Market). Под кодом Python описано, как настроить sendMail.
Технически, ограничений нет. Вы можете ввести команду в чат ролевой игры, и эта программа отправит вам все, что угодно).
Код:
# ServerCommandManager.py
# Watches World Chat (type 15). Commands like: "tp Heidel" -> send mail with mapped item.
# Requires: pip install pyodbc
import time
import threading
from datetime import datetime, timedelta
import pyodbc
import re
# ========= CONFIG =========
SQL_SERVER = r"localhost" # change if remote
SQL_DRIVER = r"ODBC Driver 17 for SQL Server" # or 18/11 as installed
USE_TRUSTED = True
SQL_USER = "CrimsonDesert"
SQL_PASS = "BlackNo.1Game#^%"
DB_LOG = "PF_BETA_LOGDB_0001"
DB_WORLD = "SA_BETA_WORLDDB_0002"
POLL_SEC = 0.5
MAIL_ENCHANT = 0
MAIL_TITLE = "Command Mail Express"
MAIL_BODY = "Mail sent from the command systems."
# Command library: city keyword (lowercase) -> itemKey (count is always 1)
TP_ITEMS = {
"velia": 43230,
"heidel": 43231,
"grana": 43232,
"altinova": 43233,
"duvencrune": 43234,
"aakman": 42491,
"crescent": 42446,
"hasrah": 43297,
"hystria": 42492,
"pirate": 42439,
"sycraia": 42265,
"valencia": 42410
# etc.
}
# ========= SQL HELPERS =========
def make_conn(dbname: str) -> pyodbc.Connection:
if USE_TRUSTED:
cs = f"DRIVER={{{SQL_DRIVER}}};SERVER={SQL_SERVER};DATABASE={dbname};Trusted_Connection=yes"
else:
cs = f"DRIVER={{{SQL_DRIVER}}};SERVER={SQL_SERVER};DATABASE={dbname};UID={SQL_USER};PWD={SQL_PASS}"
return pyodbc.connect(cs, autocommit=True)
# ========= WATCHER =========
class ChatCommandWatcher:
CMD_TP_RE = re.compile(r"^\s*tp\s+([A-Za-z]+)\s*$", re.IGNORECASE)
CMD_RESCUE_RE = re.compile(r"^\s*rescue\s*$", re.IGNORECASE)
def __init__(self):
self.stop = threading.Event()
self.last_seen_ts = datetime.utcnow() - timedelta(milliseconds=500)
self.seen = set() # (ts, userNo, msg)
self.cooldown = {} # userNo -> datetime of last tp
self.COOLDOWN_SEC = 5 * 60
self.rescue_cd = {} # for rescue
self.RESCUE_COOLDOWN_SEC = 5 * 60
self.cn_log = make_conn(DB_LOG)
self.cn_world = make_conn(DB_WORLD)
def run(self):
print("Listening for Roleplay Chat (type=15): commands 'tp <city>' and 'rescue'...")
try:
while not self.stop.is_set():
self.tick()
time.sleep(POLL_SEC)
finally:
self.close()
def tick(self):
q = f"""
SELECT TOP (200)
_registerDate, _chatMessage, _characterName, _characterNo, _userNo
FROM {DB_LOG}.PaGamePrivate.TblChatLog_INMEM WITH (READUNCOMMITTED)
WHERE _chatType = 15
AND _registerDate >= ?
ORDER BY _registerDate ASC;
"""
try:
cur = self.cn_log.cursor()
cur.execute(q, self.last_seen_ts)
rows = cur.fetchall()
except Exception as e:
print(f"[ERR] chat query failed: {e}")
return
newest_ts = None
for ts, msg, char_name, char_no, user_no in rows:
key = (ts, int(user_no or 0), str(msg or ""))
if key in self.seen:
continue
self.seen.add(key)
if msg:
self.process_message(ts, str(msg), int(user_no or 0), char_name, char_no)
if newest_ts is None or (ts and ts > newest_ts):
newest_ts = ts
if newest_ts and newest_ts >= self.last_seen_ts:
self.last_seen_ts = newest_ts + timedelta(milliseconds=1)
def process_message(self, ts, msg, user_no, char_name, char_no):
uid = int(user_no or 0)
now = datetime.utcnow()
# --- rescue (5‑min cooldown) ---
if self.CMD_RESCUE_RE.match(msg):
last = self.rescue_cd.get(uid)
if last and (now - last).total_seconds() < self.RESCUE_COOLDOWN_SEC:
remain = int(self.RESCUE_COOLDOWN_SEC - (now - last).total_seconds())
print(f"[CD] user {uid} rescue blocked: {remain}s remaining")
return
family = self.lookup_family_name_by_userno(uid)
if not family:
print(f"[WARN] rescue: cannot resolve family for userNo={uid}")
return
symno = self.send_mail(family, 43230, 1) # Map to Velia
if symno:
self.rescue_cd[uid] = now
print(f"[OK] {ts} rescue -> mailed Velia map (42563) to @{family} symNo={symno}")
else:
print(f"[ERR] sendMail failed for @{family} (rescue)")
return
# --- tp <city> (5‑min cooldown) ---
m = self.CMD_TP_RE.match(msg)
if not m:
return
city = m.group(1).lower()
last = self.cooldown.get(uid)
if last and (now - last).total_seconds() < self.COOLDOWN_SEC:
remain = int(self.COOLDOWN_SEC - (now - last).total_seconds())
print(f"[CD] user {uid} tp '{city}' blocked: {remain}s remaining")
return
item_key = TP_ITEMS.get(city)
if not item_key:
print(f"[INFO] unknown tp city '{city}' from user {uid} ({char_name}); ignore")
return
family = self.lookup_family_name_by_userno(uid)
if not family:
print(f"[WARN] tp {city}: cannot resolve family for userNo={uid}")
return
symno = self.send_mail(family, item_key, 1)
if symno:
self.cooldown[uid] = now # or rescue_cd
print(f"[OK] {ts} tp {city} -> mailed item {item_key} x1 to @{family} (symNo={symno})")
else:
print(f"[ERR] sendMail failed for @{family} (tp {city})")
def lookup_family_name_by_userno(self, user_no) -> str:
q = f"""
SELECT TOP (1) _userNickname
FROM {DB_WORLD}.PaGamePrivate.TblUserInformation WITH (READUNCOMMITTED)
WHERE _userNo = ?;
"""
try:
cur = self.cn_world.cursor()
cur.execute(q, user_no)
row = cur.fetchone()
return row[0] if row else None
except Exception as e:
print(f"[ERR] family lookup failed: {e}")
return None
def send_mail(self, to_family: str, item_key: int, count: int) -> str:
try:
cur = self.cn_world.cursor()
cur.execute("""
DECLARE @symNo NVARCHAR(50);
EXEC dbo.sendMail
@toFamilyName = ?,
@itemKey = ?,
@itemCount = ?,
@enchant = ?,
@title = ?,
@contents = ?,
@symNo = @symNo OUTPUT;
SELECT @symNo AS symNo;
""", (to_family, item_key, count, MAIL_ENCHANT, MAIL_TITLE, MAIL_BODY))
sym_no = None
# Walk all result sets; pick the one that exposes symNo
while True:
if cur.description:
cols = [d[0].lower() for d in cur.description]
if "symno" in cols:
idx = cols.index("symno")
row = cur.fetchone()
if row:
sym_no = row[idx]
# drain remaining rows in this set
for _ in cur:
pass
if not cur.nextset():
break
return sym_no or "OK" # treat as success even if symNo wasn’t surfaced
except Exception as e:
print(f"[ERR] sendMail error: {e}")
return None
def main():
w = ChatCommandWatcher()
try:
w.run()
except KeyboardInterrupt:
w.stop.set()
print("Stopping…")
if __name__ == "__main__":
main()