125 lines
4.1 KiB
Python
125 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Парсер en-GB HTM файлов ElsaWin: hs2 + www"""
|
|
import os, re, json, sys
|
|
from pathlib import Path
|
|
|
|
def parse_htm_file(path):
|
|
"""Парсить один HTM-файл в UTF-16LE."""
|
|
with open(path, "rb") as f:
|
|
raw = f.read()
|
|
|
|
# Проверяем кодировку
|
|
if raw[:2] == b"\xff\xfe":
|
|
text = raw.decode("utf-16-le", errors="replace")
|
|
elif raw[:2] == b"\xfe\xff":
|
|
text = raw.decode("utf-16-be", errors="replace")
|
|
else:
|
|
text = raw.decode("ascii", errors="replace")
|
|
|
|
# Title
|
|
m = re.search(r"<title>(.*?)</title>", text, re.IGNORECASE | re.DOTALL)
|
|
title = m.group(1).strip() if m else ""
|
|
|
|
# Body text
|
|
body = re.sub(r"<script[^>]*>.*?</script>", " ", text, flags=re.IGNORECASE | re.DOTALL)
|
|
body = re.sub(r"<style[^>]*>.*?</style>", " ", body, flags=re.IGNORECASE | re.DOTALL)
|
|
body = re.sub(r"<[^>]+>", " ", body)
|
|
body = re.sub(r" ", " ", body)
|
|
body = re.sub(r"&\w+;", " ", body)
|
|
body = re.sub(r"\s+", " ", body).strip()
|
|
|
|
return {"title": title, "text": body}
|
|
|
|
|
|
def main():
|
|
if len(sys.argv) < 3:
|
|
print(f"Usage: {sys.argv[0]} <elsa_docs_dir> <output_dir>")
|
|
sys.exit(1)
|
|
|
|
elsa_docs = os.path.abspath(sys.argv[1])
|
|
out_dir = os.path.abspath(sys.argv[2])
|
|
os.makedirs(out_dir, exist_ok=True)
|
|
|
|
en_gb_dirs = [
|
|
# hs2 — in nested structure: hs2/V/en-GB/src/XX/XXXXXX/master.htm
|
|
os.path.join(elsa_docs, "hs2", "V", "en-GB"),
|
|
# www — various paths under www/.../en-GB/
|
|
]
|
|
|
|
# Find all en-GB HTM
|
|
all_files = []
|
|
for root, dirs, files in os.walk(elsa_docs):
|
|
if "/en-GB/" in root and not "/slp/" in root:
|
|
for f in files:
|
|
if f.endswith(".htm") or f.endswith(".html"):
|
|
all_files.append(os.path.join(root, f))
|
|
|
|
# Skip already-known-malformed (slp — machine code, already excluded above)
|
|
# Only au, hs2, www, igg
|
|
good_files = [f for f in all_files if any(d in f for d in ["/www/", "/hs2/", "/au/"])]
|
|
|
|
print(f"Found {len(good_files)} en-GB HTM files (excluding slp)")
|
|
|
|
# Split by source
|
|
by_source = {}
|
|
for f in good_files:
|
|
if "/au/" in f:
|
|
src = "au"
|
|
elif "/hs2/" in f:
|
|
src = "hs2"
|
|
elif "/www/" in f:
|
|
src = "www"
|
|
else:
|
|
src = "other"
|
|
by_source.setdefault(src, []).append(f)
|
|
|
|
for src, files in by_source.items():
|
|
print(f" {src}: {len(files)} files")
|
|
|
|
# Parse and save
|
|
total = 0
|
|
part_size = 5000
|
|
part_idx = 0
|
|
buf = []
|
|
out_file = None
|
|
|
|
for src, files in by_source.items():
|
|
for fpath in sorted(files):
|
|
try:
|
|
doc = parse_htm_file(fpath)
|
|
rel = os.path.relpath(fpath, elsa_docs)
|
|
record = {
|
|
"source": f"elsa_{src}",
|
|
"file": "/docs/" + rel,
|
|
"title": doc["title"],
|
|
"text": doc["text"],
|
|
}
|
|
buf.append(record)
|
|
total += 1
|
|
|
|
if len(buf) >= part_size:
|
|
out_file = os.path.join(out_dir, f"part_{part_idx:04d}.jsonl")
|
|
with open(out_file, "w", encoding="utf-8") as f:
|
|
for r in buf:
|
|
f.write(json.dumps(r, ensure_ascii=False) + "\n")
|
|
print(f" Part {part_idx:04d}: {len(buf)} records → {out_file}")
|
|
buf = []
|
|
part_idx += 1
|
|
except Exception as e:
|
|
print(f" ERROR {fpath}: {e}", file=sys.stderr)
|
|
|
|
# Flush remaining
|
|
if buf:
|
|
out_file = os.path.join(out_dir, f"part_{part_idx:04d}.jsonl")
|
|
with open(out_file, "w", encoding="utf-8") as f:
|
|
for r in buf:
|
|
f.write(json.dumps(r, ensure_ascii=False) + "\n")
|
|
print(f" Part {part_idx:04d}: {len(buf)} records → {out_file}")
|
|
part_idx += 1
|
|
|
|
print(f"\nTotal: {total} documents in {part_idx} parts → {out_dir}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|