Alex
1e361301ad
- Add search relevance scoring: exact title > title prefix > title contains > Core Rulebook source > content match - Add DB_PASSWORD env var to pm2.config.js so server can connect to MariaDB - Add update-rules-from-sources.py: imports 45 skills from CSV and 44 talents from CR.txt, removes garbage advance-table entries - Add fix-talent-titles.py: normalises OCR-garbled talent titles, removes 28 duplicates, fixes 65 entries - Add cleanup-skill-dupes.py: removes duplicate skills from csv-import/fandom/sanitized sources, recategorises 55 lore entries Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
366 lines
15 KiB
Python
366 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Update rules database from authoritative sources:
|
||
1. Skills → database/deathwatch_skills_p94_107.csv
|
||
2. Talents → database/rules/CR.txt (Chapter IV, lines ~7272-8431)
|
||
3. Delete garbled advance-table entries
|
||
|
||
Uses the REST API at http://localhost:5000
|
||
"""
|
||
|
||
import csv, json, re, sys, urllib.request, urllib.error
|
||
from pathlib import Path
|
||
|
||
REPO = Path(__file__).resolve().parent.parent
|
||
API = "http://localhost:5000/api"
|
||
GM_SECRET = "bongo"
|
||
|
||
# ── helpers ───────────────────────────────────────────────────────────────────
|
||
|
||
def api_call(method, path, body=None):
|
||
url = API + path
|
||
data = json.dumps(body).encode() if body else None
|
||
headers = {"Content-Type": "application/json", "x-gm-secret": GM_SECRET}
|
||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=15) as r:
|
||
return json.loads(r.read())
|
||
except urllib.error.HTTPError as e:
|
||
print(f" HTTP {e.code} {method} {path}: {e.read().decode()[:200]}")
|
||
return None
|
||
|
||
def upsert_rule(rule_id, title, content, page, source, source_abbr, category):
|
||
"""Insert or replace a rule in the DB via direct SQL through a helper endpoint.
|
||
Falls back to delete+insert since the API doesn't expose a PATCH endpoint."""
|
||
import mysql.connector
|
||
# We'll do this via mysql directly (faster than HTTP for bulk)
|
||
return (rule_id, title, content, page, source, source_abbr, category)
|
||
|
||
# ── 1. Skills from CSV ─────────────────────────────────────────────────────────
|
||
|
||
def load_skills():
|
||
csvpath = REPO / "database" / "deathwatch_skills_p94_107.csv"
|
||
with open(csvpath, newline="", encoding="utf-8") as f:
|
||
return list(csv.DictReader(f))
|
||
|
||
def build_skill_content(row):
|
||
parts = []
|
||
if row.get("skill_text"):
|
||
parts.append(row["skill_text"].strip())
|
||
if row.get("skill_description"):
|
||
parts.append(row["skill_description"].strip())
|
||
if row.get("skill_use"):
|
||
parts.append(f"Use: {row['skill_use'].strip()}")
|
||
return "\n\n".join(parts)
|
||
|
||
# ── 2. Talents from CR.txt ─────────────────────────────────────────────────────
|
||
|
||
# Known talent names (from Table 4-1, p108-111 of Core Rulebook)
|
||
KNOWN_TALENTS = {
|
||
"abhor the witch", "air of authority", "ambidextrous",
|
||
"assassin strike", "astartes weapon specialisation",
|
||
"battle rage", "berserk charge", "binary chatter",
|
||
"blademaster", "blind fighting", "bulging biceps",
|
||
"catfall", "combat formation", "combat master",
|
||
"crack shot", "crushing blow", "deadeye shot",
|
||
"die hard", "disturbing voice", "double team",
|
||
"dual shot", "dual strike", "enemy", "eye of vengeance",
|
||
"fast hands", "fearless", "foresight",
|
||
"gunslinger", "hard target", "hatred",
|
||
"heightened senses", "hip shooting", "horde fighter",
|
||
"hunter of aliens", "independent targeting",
|
||
"inspire wrath", "into the jaws of hell",
|
||
"iron affinity", "iron discipline", "iron jaw",
|
||
"jaded", "killing strike", "leap up",
|
||
"leaping dodge", "light sleeper", "lightning attack",
|
||
"lightning reflexes", "litany of hate", "logis implant",
|
||
"luminen blast", "luminen charge", "luminen shock",
|
||
"maglev grace", "marksman", "master craftsman",
|
||
"mechadendrite use", "meditation", "mental rage",
|
||
"mighty shot", "nerves of steel", "night vision",
|
||
"peer", "pistol training", "precise blow",
|
||
"psy rating", "quick draw", "rapid reaction",
|
||
"rapid reload", "resistance", "rite of fear",
|
||
"rite of pure thought", "sacrificial strike",
|
||
"sharpshooter", "slayer of daemons", "sound constitution",
|
||
"sprint", "step aside", "storm of iron",
|
||
"strong minded", "swift attack", "talent",
|
||
"thunder charge", "true grit", "two weapon wielder",
|
||
"unarmed master", "unarmed warrior",
|
||
"unnatural characteristic", "warp affinity",
|
||
"weapon training",
|
||
# Chapter speciality talents
|
||
"astartes assault", "astartes devastator", "astartes librarian",
|
||
"astartes tactical", "astartes techmarine",
|
||
"basic weapon training", "bolt weapon mastery",
|
||
"energy cache", "enhanced bionic frame",
|
||
"frenzy", "furious assault", "flesh render",
|
||
"grenade mastery", "hammer blow", "hard target",
|
||
"head shot", "incendiary charge", "mechadendrite use",
|
||
"overcharge", "overwatch", "precision killer",
|
||
"stay vigilant", "tactical advance",
|
||
}
|
||
|
||
def normalise_talent_name(raw):
|
||
"""Normalise OCR talent name to lowercase clean string."""
|
||
# Remove OCR random-cap artifacts: if string is mixed case with no obvious pattern,
|
||
# normalise to lowercase
|
||
return re.sub(r'\s+', ' ', raw).strip().lower()
|
||
|
||
def split_two_columns(line):
|
||
"""Split a line from a two-column PDF layout.
|
||
Returns (left, right) stripped strings, either may be empty."""
|
||
# Find the largest run of whitespace (gap between columns)
|
||
matches = list(re.finditer(r' {5,}', line))
|
||
if not matches:
|
||
return line.strip(), ""
|
||
# Pick the widest gap; if there are multiple, prefer the one closest to centre
|
||
best = max(matches, key=lambda m: m.end() - m.start())
|
||
left = line[:best.start()].strip()
|
||
right = line[best.end():].strip()
|
||
return left, right
|
||
|
||
def looks_like_talent_name(s):
|
||
"""Heuristic: a talent name is a short phrase (≤8 words) with no digits/punctuation."""
|
||
s = s.strip()
|
||
if not s or len(s) > 70 or len(s) < 3:
|
||
return False
|
||
if re.search(r'[0-9:;|]', s):
|
||
return False
|
||
words = s.split()
|
||
if len(words) > 8:
|
||
return False
|
||
# Must be mostly letters
|
||
letters = sum(c.isalpha() for c in s)
|
||
if letters / len(s) < 0.7:
|
||
return False
|
||
norm = normalise_talent_name(s)
|
||
# Must match a known talent or be a plausible name
|
||
for kt in KNOWN_TALENTS:
|
||
if kt == norm or kt.startswith(norm) or norm.startswith(kt):
|
||
return True
|
||
# fallback: allow if it looks like a heading (no verb, short)
|
||
return False
|
||
|
||
def parse_talents_from_cr(cr_path):
|
||
"""Parse the talent descriptions from CR.txt using column-splitting."""
|
||
with open(cr_path, encoding="utf-8") as f:
|
||
lines = f.readlines()
|
||
|
||
# Find start of talent descriptions (after the talent table)
|
||
START_HINT = 7272 # from analysis
|
||
END_HINT = 8450
|
||
|
||
relevant = lines[START_HINT - 1 : END_HINT]
|
||
|
||
# Rebuild two separate column streams
|
||
left_stream = []
|
||
right_stream = []
|
||
|
||
for line in relevant:
|
||
line = line.rstrip('\n')
|
||
l, r = split_two_columns(line)
|
||
left_stream.append(l)
|
||
right_stream.append(r)
|
||
|
||
def extract_talents_from_stream(stream):
|
||
"""Extract {name, prerequisites, description} dicts from one column stream."""
|
||
talents = []
|
||
current = None
|
||
prereq_done = False
|
||
|
||
for line in stream:
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
|
||
# Check if this is a "Prerequisites:" line
|
||
pm = re.match(r'^Prerequisites?:\s*(.*)', line, re.IGNORECASE)
|
||
if pm and current is not None:
|
||
current['prerequisites'] = pm.group(1).strip()
|
||
prereq_done = True
|
||
continue
|
||
|
||
# Check if it looks like a talent name
|
||
if looks_like_talent_name(line) and (current is None or prereq_done):
|
||
if current and current.get('description'):
|
||
talents.append(current)
|
||
current = {'name': line, 'prerequisites': '—', 'description': ''}
|
||
prereq_done = False
|
||
continue
|
||
|
||
# Otherwise it's description text
|
||
if current is not None:
|
||
if current['description']:
|
||
current['description'] += ' ' + line
|
||
else:
|
||
current['description'] = line
|
||
|
||
if current and current.get('description'):
|
||
talents.append(current)
|
||
|
||
return talents
|
||
|
||
left_talents = extract_talents_from_stream(left_stream)
|
||
right_talents = extract_talents_from_stream(right_stream)
|
||
|
||
# Merge and deduplicate by normalised name
|
||
all_talents = {}
|
||
for t in left_talents + right_talents:
|
||
norm = normalise_talent_name(t['name'])
|
||
if norm not in all_talents or len(t['description']) > len(all_talents[norm]['description']):
|
||
all_talents[norm] = t
|
||
|
||
return list(all_talents.values())
|
||
|
||
# ── 3. Direct DB update via mysql.connector ────────────────────────────────────
|
||
|
||
def get_db():
|
||
import mysql.connector
|
||
return mysql.connector.connect(
|
||
host="192.168.1.113", port=3307,
|
||
user="deathwatch",
|
||
password=open(REPO / ".env.db").read().strip() if (REPO / ".env.db").exists() else "DwRoller@2025!",
|
||
database="deathwatch"
|
||
)
|
||
|
||
def run():
|
||
print("=== Deathwatch Rules Updater ===\n")
|
||
|
||
# Try direct DB connection
|
||
try:
|
||
db = get_db()
|
||
cursor = db.cursor()
|
||
print("Connected to MariaDB directly.\n")
|
||
except Exception as e:
|
||
print(f"Cannot connect to MariaDB: {e}")
|
||
print("Falling back to HTTP API (slower).")
|
||
db = None
|
||
cursor = None
|
||
|
||
def exec_sql(sql, params=()):
|
||
if cursor:
|
||
cursor.execute(sql, params)
|
||
else:
|
||
print(f" [no-db] would execute: {sql[:80]}")
|
||
|
||
def upsert(rule_id, title, content, page, source, source_abbr, category):
|
||
if cursor:
|
||
cursor.execute(
|
||
"""INSERT INTO rules (rule_id, title, content, page, source, source_abbr, category)
|
||
VALUES (%s, %s, %s, %s, %s, %s, %s)
|
||
ON DUPLICATE KEY UPDATE
|
||
title=VALUES(title), content=VALUES(content),
|
||
page=VALUES(page), source=VALUES(source),
|
||
source_abbr=VALUES(source_abbr), category=VALUES(category)""",
|
||
(rule_id, title, content, page, source, source_abbr, category)
|
||
)
|
||
else:
|
||
print(f" [no-db] upsert '{title}'")
|
||
|
||
# ── Step 1: Delete garbage advance-table entries ──────────────────────────
|
||
print("Step 1: Removing garbled advance-table entries...")
|
||
garbage_patterns = [
|
||
# Advance tables masquerading as rules
|
||
"title LIKE 'AdvanceCostType%'",
|
||
# Rules whose entire content is a cost/prerequisites table
|
||
"content REGEXP '^[A-Za-z ]+[[:space:]]+[0-9,]+[[:space:]]+(Skill|Talent)[[:space:]]'",
|
||
# Very short meaningless entries (under 30 chars content, sanitized source)
|
||
"(source = 'sanitized' AND LENGTH(content) < 40)",
|
||
]
|
||
deleted = 0
|
||
for pat in garbage_patterns:
|
||
if cursor:
|
||
cursor.execute(f"SELECT COUNT(*) FROM rules WHERE {pat}")
|
||
count = cursor.fetchone()[0]
|
||
cursor.execute(f"DELETE FROM rules WHERE {pat}")
|
||
print(f" Deleted {count} rows matching: {pat[:60]}")
|
||
deleted += count
|
||
else:
|
||
print(f" [no-db] would delete: {pat[:60]}")
|
||
print(f" Total deleted: {deleted}\n")
|
||
|
||
# ── Step 2: Skills from CSV ───────────────────────────────────────────────
|
||
print("Step 2: Upserting skills from CSV...")
|
||
skills = load_skills()
|
||
for skill in skills:
|
||
name = skill['name'].strip()
|
||
content = build_skill_content(skill)
|
||
char = skill.get('characteristic', '').strip()
|
||
stype = skill.get('type', '').strip()
|
||
desc_header = f"{stype} Skill – {char}" if char else stype
|
||
full_content = f"{desc_header}\n\n{content}" if desc_header else content
|
||
|
||
rule_id = "skill_" + re.sub(r'[^a-z0-9]', '_', name.lower())
|
||
upsert(rule_id, name, full_content, None,
|
||
"Core Rulebook", "CR", "skills")
|
||
print(f" ✓ {name}")
|
||
if db:
|
||
db.commit()
|
||
print(f" {len(skills)} skills updated.\n")
|
||
|
||
# ── Step 3: Talents from CR.txt ───────────────────────────────────────────
|
||
print("Step 3: Parsing and upserting talents from CR.txt...")
|
||
cr_path = REPO / "database" / "rules" / "CR.txt"
|
||
talents = parse_talents_from_cr(cr_path)
|
||
print(f" Parsed {len(talents)} talent entries from CR.txt")
|
||
|
||
imported = 0
|
||
for t in talents:
|
||
name = re.sub(r'\s+', ' ', t['name']).strip()
|
||
# Clean up OCR capitalisation: convert random-caps to title case
|
||
name_clean = re.sub(r'([A-Z])', lambda m: m.group(1).lower(), name)
|
||
name_clean = ' '.join(w.capitalize() for w in name_clean.split())
|
||
|
||
prereq = t.get('prerequisites', '—').strip()
|
||
desc = t.get('description', '').strip()
|
||
if len(desc) < 20:
|
||
continue # Skip empty/garbage entries
|
||
|
||
content = f"Prerequisites: {prereq}\n\n{desc}"
|
||
rule_id = "talent_" + re.sub(r'[^a-z0-9]', '_', name_clean.lower())
|
||
|
||
upsert(rule_id, name_clean, content, None,
|
||
"Core Rulebook", "CR", "talents")
|
||
imported += 1
|
||
|
||
if db:
|
||
db.commit()
|
||
print(f" {imported} talents upserted.\n")
|
||
|
||
# ── Step 4: Fix existing sanitized rules – remove pure table entries ──────
|
||
print("Step 4: Cleaning up remaining garbled sanitized entries...")
|
||
if cursor:
|
||
# Rules where the title IS the first line of the content (typical for advance tables)
|
||
cursor.execute("""
|
||
DELETE FROM rules
|
||
WHERE source = 'sanitized'
|
||
AND (
|
||
content LIKE '%Cost%Type%Prerequisites%'
|
||
OR content REGEXP '[0-9]{3,}[[:space:]]+(Skill|Talent)[[:space:]]'
|
||
OR (title = content)
|
||
)
|
||
""")
|
||
n = cursor.rowcount
|
||
db.commit()
|
||
print(f" Deleted {n} more garbled entries.\n")
|
||
else:
|
||
print(" [no-db] skipped.\n")
|
||
|
||
# ── Done ──────────────────────────────────────────────────────────────────
|
||
if cursor:
|
||
cursor.execute("SELECT COUNT(*) FROM rules")
|
||
total = cursor.fetchone()[0]
|
||
cursor.execute("SELECT source, COUNT(*) FROM rules GROUP BY source ORDER BY COUNT(*) DESC")
|
||
by_source = cursor.fetchall()
|
||
print(f"=== Done. Total rules in DB: {total} ===")
|
||
for src, cnt in by_source:
|
||
print(f" {src}: {cnt}")
|
||
cursor.close()
|
||
db.close()
|
||
else:
|
||
print("=== Done (dry-run, no DB connection). ===")
|
||
|
||
if __name__ == "__main__":
|
||
run()
|