I wanted to be able to teleport or use commands and stuff in game.
I found that using the role-play channel, chat is logged.
I found a way to read the log and react based on that.
Currently, I have it set up where you can execute a few commands:
rescue = Sends a teleport item for Velia
tp velia = Sends a teleport item for Velia
tp followed by: heidel, grana, altinova, aakman, Duvencrune, crescent, harsh, hystria, pirate, sycraia, valencia
Will send you an item, via mail, that will teleport you to those locations. Its quite nice.
1. You will need to install python. I'm using 3.9.5 because its installed with Corsair.
2. You will need sendmail (in this same forum, I have a post about Central/Market, Black Market. Under the python code, is how to set up sendMail.
Technically, the sky is the limit. You can enter a command into role-play chat and have this program send you whatever.
I found that using the role-play channel, chat is logged.
I found a way to read the log and react based on that.
Currently, I have it set up where you can execute a few commands:
rescue = Sends a teleport item for Velia
tp velia = Sends a teleport item for Velia
tp followed by: heidel, grana, altinova, aakman, Duvencrune, crescent, harsh, hystria, pirate, sycraia, valencia
Will send you an item, via mail, that will teleport you to those locations. Its quite nice.
1. You will need to install python. I'm using 3.9.5 because its installed with Corsair.
2. You will need sendmail (in this same forum, I have a post about Central/Market, Black Market. Under the python code, is how to set up sendMail.
Technically, the sky is the limit. You can enter a command into role-play chat and have this program send you whatever.
Python:
# 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()
