from __future__ import annotations
|
|
import argparse
|
import csv
|
import getpass
|
import hashlib
|
import json
|
import os
|
import re
|
from datetime import datetime
|
from pathlib import Path
|
|
import pymysql
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
DEFAULT_OUT_DIR = ROOT / "data" / "legacy_case_import_20260624"
|
IMPORT_BATCH_ID = "DL_LEGACY_CASE_IMPORT_20260624_V1"
|
SOURCE_NAME = "legacy_external_information_casebook"
|
SOURCE_KIND = "markdown_casebook"
|
|
|
def project_rel(path: Path) -> str:
|
try:
|
return str(path.resolve().relative_to(ROOT)).replace("\\", "/")
|
except ValueError:
|
return f"external_source/{path.name}"
|
|
|
def sha256_text(text: str) -> str:
|
return hashlib.sha256(text.encode("utf-8")).hexdigest()
|
|
|
def sha256_file(path: Path) -> str:
|
h = hashlib.sha256()
|
with path.open("rb") as f:
|
for chunk in iter(lambda: f.read(1024 * 1024), b""):
|
h.update(chunk)
|
return h.hexdigest()
|
|
|
def infer_level(body: str) -> str:
|
levels = []
|
for level in ("L1", "L2", "L3"):
|
if re.search(rf"(?<![A-Z0-9]){level}(?![A-Z0-9])", body):
|
levels.append(level)
|
return "+".join(levels) if levels else "UNKNOWN"
|
|
|
def infer_case_family(title: str, body: str) -> str:
|
text = title + "\n" + body[:2000]
|
rules = [
|
("AUTO_BATCH", r"自动批量案例|AUTOBATCH"),
|
("COUNTERINTUITIVE_KLINE", r"反常识|不跌反涨|坏标题|利空.*上涨"),
|
("STATE_OWNED_REPURCHASE", r"招商局|回购|增持"),
|
("REGULATORY_RISK", r"立案|证监会|监管|违规|实控人|关键人"),
|
("AUDIT_REMEDIATION", r"审计|非标|影响消除|保留意见"),
|
("DISTRESSED_RESTRUCTURING", r"重整|ST|退市|债务|司法拍卖|壳"),
|
("INDUSTRIAL_GROWTH", r"机器人|商业航天|AI|订单|扩产|股权激励"),
|
]
|
for family, pattern in rules:
|
if re.search(pattern, text, flags=re.I):
|
return family
|
return "GENERAL_DARKLINE_CASE"
|
|
|
def split_cases(source_text: str, source_rel: str, body_dir: Path) -> list[dict[str, str]]:
|
heading_pattern = re.compile(r"^##\s+(.+)$", re.M)
|
matches = list(heading_pattern.finditer(source_text))
|
cases: list[dict[str, str]] = []
|
case_no = 0
|
|
for i, match in enumerate(matches):
|
title = match.group(1).strip()
|
start = match.start()
|
end = matches[i + 1].start() if i + 1 < len(matches) else len(source_text)
|
body = source_text[start:end].strip()
|
|
is_case = (
|
title.startswith("案例")
|
or title.startswith("自动批量案例")
|
or title.startswith("DARKLINE-CASE")
|
)
|
if not is_case or title.startswith("案例 X"):
|
continue
|
|
case_no += 1
|
case_id = f"DLCASE_LEGACY_{case_no:04d}"
|
body_hash = sha256_text(body)
|
body_path = body_dir / f"{case_id}.md"
|
body_path.write_text(body + "\n", encoding="utf-8")
|
|
cases.append(
|
{
|
"case_id": case_id,
|
"import_batch_id": IMPORT_BATCH_ID,
|
"case_order": str(case_no),
|
"case_title": title,
|
"case_family": infer_case_family(title, body),
|
"case_level": infer_level(body),
|
"current_status": "LEGACY_IMPORTED_REVIEW",
|
"case_body_hash": body_hash,
|
"local_body_path": project_rel(body_path),
|
"source_document_path": source_rel,
|
"source_section_heading": f"## {title}",
|
}
|
)
|
|
return cases
|
|
|
def write_csv(path: Path, rows: list[dict[str, str]], fields: list[str]) -> None:
|
with path.open("w", newline="", encoding="utf-8-sig") as f:
|
writer = csv.DictWriter(f, fieldnames=fields)
|
writer.writeheader()
|
writer.writerows(rows)
|
|
|
def write_manifest(manifest_csv: Path, paths: list[Path]) -> None:
|
rows = []
|
for path in sorted(paths):
|
rows.append(
|
{
|
"artifact_path": project_rel(path),
|
"artifact_name": path.name,
|
"artifact_type": path.suffix.lstrip(".") or "directory",
|
"size_bytes": path.stat().st_size,
|
"sha256": sha256_file(path),
|
"artifact_status": "READY",
|
}
|
)
|
write_csv(
|
manifest_csv,
|
rows,
|
[
|
"artifact_path",
|
"artifact_name",
|
"artifact_type",
|
"size_bytes",
|
"sha256",
|
"artifact_status",
|
],
|
)
|
|
|
def get_password(args: argparse.Namespace) -> str:
|
if args.password:
|
return args.password
|
if os.environ.get("DARKLINE_MYSQL_PASSWORD"):
|
return os.environ["DARKLINE_MYSQL_PASSWORD"]
|
if os.environ.get("TIANXIA_MYSQL_PASSWORD"):
|
return os.environ["TIANXIA_MYSQL_PASSWORD"]
|
return getpass.getpass("MySQL password: ")
|
|
|
def connect_mysql(args: argparse.Namespace):
|
return pymysql.connect(
|
host=args.host,
|
port=args.port,
|
user=args.user,
|
password=get_password(args),
|
database=args.database,
|
charset="utf8mb4",
|
autocommit=True,
|
)
|
|
|
def import_mysql(
|
args: argparse.Namespace,
|
cases: list[dict[str, str]],
|
source_hash: str,
|
source_rel: str,
|
out_dir: Path,
|
) -> dict[str, object]:
|
imported_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
conn = connect_mysql(args)
|
with conn.cursor() as cur:
|
cur.execute(
|
"""
|
CREATE TABLE IF NOT EXISTS dl_case_import_batch (
|
import_batch_id VARCHAR(96) PRIMARY KEY,
|
source_name VARCHAR(128) NOT NULL,
|
source_kind VARCHAR(64) NOT NULL,
|
source_document_path VARCHAR(512) NOT NULL,
|
source_hash CHAR(64) NOT NULL,
|
case_count INT NOT NULL,
|
data_dir VARCHAR(512) NOT NULL,
|
imported_at DATETIME NOT NULL,
|
note VARCHAR(512) NULL
|
) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci
|
"""
|
)
|
cur.execute(
|
"""
|
CREATE TABLE IF NOT EXISTS dl_case_record (
|
case_id VARCHAR(96) PRIMARY KEY,
|
import_batch_id VARCHAR(96) NOT NULL,
|
case_order INT NOT NULL,
|
case_title VARCHAR(512) NOT NULL,
|
case_family VARCHAR(96) NOT NULL,
|
case_level VARCHAR(32) NOT NULL,
|
current_status VARCHAR(64) NOT NULL,
|
case_body_md LONGTEXT NOT NULL,
|
case_body_hash CHAR(64) NOT NULL,
|
local_body_path VARCHAR(512) NOT NULL,
|
source_document_path VARCHAR(512) NOT NULL,
|
source_section_heading VARCHAR(512) NOT NULL,
|
created_at DATETIME NOT NULL,
|
updated_at DATETIME NOT NULL,
|
KEY idx_dl_case_batch (import_batch_id),
|
KEY idx_dl_case_family (case_family),
|
KEY idx_dl_case_level (case_level)
|
) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci
|
"""
|
)
|
cur.execute(
|
"""
|
CREATE TABLE IF NOT EXISTS dl_darkline_hypothesis (
|
darkline_hypothesis_id VARCHAR(96) PRIMARY KEY,
|
case_id VARCHAR(96) NOT NULL,
|
hypothesis_title VARCHAR(512) NOT NULL,
|
darkline_level VARCHAR(32) NOT NULL,
|
hypothesis_family VARCHAR(96) NOT NULL,
|
hypothesis_status VARCHAR(64) NOT NULL,
|
source_case_body_hash CHAR(64) NOT NULL,
|
created_at DATETIME NOT NULL,
|
updated_at DATETIME NOT NULL,
|
KEY idx_dl_hypothesis_case (case_id),
|
KEY idx_dl_hypothesis_level (darkline_level)
|
) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci
|
"""
|
)
|
cur.execute(
|
"""
|
CREATE TABLE IF NOT EXISTS dl_chain_state (
|
state_record_id VARCHAR(128) PRIMARY KEY,
|
darkline_hypothesis_id VARCHAR(96) NOT NULL,
|
case_id VARCHAR(96) NOT NULL,
|
chain_clarity_status VARCHAR(64) NOT NULL,
|
reality_confirmation_status VARCHAR(64) NOT NULL,
|
market_manifestation_status VARCHAR(64) NOT NULL,
|
validation_readout_status VARCHAR(64) NOT NULL,
|
current_consumption_level VARCHAR(64) NOT NULL,
|
state_record_time DATETIME NOT NULL,
|
note VARCHAR(512) NULL,
|
KEY idx_dl_chain_case (case_id),
|
KEY idx_dl_chain_hypothesis (darkline_hypothesis_id)
|
) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci
|
"""
|
)
|
|
cur.execute(
|
"""
|
INSERT INTO dl_case_import_batch (
|
import_batch_id, source_name, source_kind, source_document_path,
|
source_hash, case_count, data_dir, imported_at, note
|
) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
ON DUPLICATE KEY UPDATE
|
source_hash=VALUES(source_hash),
|
case_count=VALUES(case_count),
|
data_dir=VALUES(data_dir),
|
imported_at=VALUES(imported_at),
|
note=VALUES(note)
|
""",
|
(
|
IMPORT_BATCH_ID,
|
SOURCE_NAME,
|
SOURCE_KIND,
|
source_rel,
|
source_hash,
|
len(cases),
|
project_rel(out_dir),
|
imported_at,
|
"Legacy casebook imported as historical review data for standalone darkline.",
|
),
|
)
|
|
for case in cases:
|
body = (ROOT / case["local_body_path"]).read_text(encoding="utf-8")
|
cur.execute(
|
"""
|
INSERT INTO dl_case_record (
|
case_id, import_batch_id, case_order, case_title, case_family,
|
case_level, current_status, case_body_md, case_body_hash,
|
local_body_path, source_document_path, source_section_heading,
|
created_at, updated_at
|
) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
ON DUPLICATE KEY UPDATE
|
case_title=VALUES(case_title),
|
case_family=VALUES(case_family),
|
case_level=VALUES(case_level),
|
current_status=VALUES(current_status),
|
case_body_md=VALUES(case_body_md),
|
case_body_hash=VALUES(case_body_hash),
|
local_body_path=VALUES(local_body_path),
|
source_document_path=VALUES(source_document_path),
|
source_section_heading=VALUES(source_section_heading),
|
updated_at=VALUES(updated_at)
|
""",
|
(
|
case["case_id"],
|
case["import_batch_id"],
|
int(case["case_order"]),
|
case["case_title"],
|
case["case_family"],
|
case["case_level"],
|
case["current_status"],
|
body,
|
case["case_body_hash"],
|
case["local_body_path"],
|
case["source_document_path"],
|
case["source_section_heading"],
|
imported_at,
|
imported_at,
|
),
|
)
|
cur.execute(
|
"""
|
INSERT INTO dl_darkline_hypothesis (
|
darkline_hypothesis_id, case_id, hypothesis_title, darkline_level,
|
hypothesis_family, hypothesis_status, source_case_body_hash,
|
created_at, updated_at
|
) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
ON DUPLICATE KEY UPDATE
|
hypothesis_title=VALUES(hypothesis_title),
|
darkline_level=VALUES(darkline_level),
|
hypothesis_family=VALUES(hypothesis_family),
|
hypothesis_status=VALUES(hypothesis_status),
|
source_case_body_hash=VALUES(source_case_body_hash),
|
updated_at=VALUES(updated_at)
|
""",
|
(
|
case["case_id"],
|
case["case_id"],
|
case["case_title"],
|
case["case_level"],
|
case["case_family"],
|
"LEGACY_IMPORTED_REVIEW",
|
case["case_body_hash"],
|
imported_at,
|
imported_at,
|
),
|
)
|
cur.execute(
|
"""
|
INSERT INTO dl_chain_state (
|
state_record_id, darkline_hypothesis_id, case_id,
|
chain_clarity_status, reality_confirmation_status,
|
market_manifestation_status, validation_readout_status,
|
current_consumption_level, state_record_time, note
|
) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
|
ON DUPLICATE KEY UPDATE
|
chain_clarity_status=VALUES(chain_clarity_status),
|
reality_confirmation_status=VALUES(reality_confirmation_status),
|
market_manifestation_status=VALUES(market_manifestation_status),
|
validation_readout_status=VALUES(validation_readout_status),
|
current_consumption_level=VALUES(current_consumption_level),
|
state_record_time=VALUES(state_record_time),
|
note=VALUES(note)
|
""",
|
(
|
f"{case['case_id']}_STATE_LEGACY_IMPORT_V1",
|
case["case_id"],
|
case["case_id"],
|
"LEGACY_IMPORTED_REVIEW",
|
"LEGACY_IMPORTED_REVIEW",
|
"LEGACY_IMPORTED_REVIEW",
|
"CASE_IMPORTED_FOR_REVIEW",
|
"CASEBOOK",
|
imported_at,
|
"Historical case imported. Re-validation is required before reuse.",
|
),
|
)
|
|
cur.execute("SELECT COUNT(*) FROM dl_case_record WHERE import_batch_id=%s", (IMPORT_BATCH_ID,))
|
db_case_count = cur.fetchone()[0]
|
|
conn.close()
|
return {
|
"mysql_import_status": "PASS",
|
"db_case_count": db_case_count,
|
}
|
|
|
def main() -> None:
|
parser = argparse.ArgumentParser(description="Import legacy markdown cases into darkline package files and MySQL.")
|
parser.add_argument("--source", required=True, help="Legacy markdown casebook path.")
|
parser.add_argument("--out-dir", default=str(DEFAULT_OUT_DIR))
|
parser.add_argument("--skip-mysql", action="store_true")
|
parser.add_argument("--host", default=os.environ.get("DARKLINE_MYSQL_HOST", "127.0.0.1"))
|
parser.add_argument("--port", type=int, default=int(os.environ.get("DARKLINE_MYSQL_PORT", "3306")))
|
parser.add_argument("--user", default=os.environ.get("DARKLINE_MYSQL_USER", "root"))
|
parser.add_argument("--password", default=None)
|
parser.add_argument("--database", default=os.environ.get("DARKLINE_MYSQL_DATABASE", "tianxia"))
|
args = parser.parse_args()
|
|
source = Path(args.source)
|
out_dir = Path(args.out_dir)
|
body_dir = out_dir / "case_body"
|
index_csv = out_dir / "legacy_case_index.csv"
|
summary_json = out_dir / "legacy_case_import_summary.json"
|
manifest_csv = out_dir / "legacy_case_import_manifest.csv"
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
body_dir.mkdir(parents=True, exist_ok=True)
|
|
source_text = source.read_text(encoding="utf-8")
|
source_hash = sha256_text(source_text)
|
source_rel = project_rel(source)
|
cases = split_cases(source_text, source_rel, body_dir)
|
|
fields = [
|
"case_id",
|
"import_batch_id",
|
"case_order",
|
"case_title",
|
"case_family",
|
"case_level",
|
"current_status",
|
"case_body_hash",
|
"local_body_path",
|
"source_document_path",
|
"source_section_heading",
|
]
|
write_csv(index_csv, cases, fields)
|
|
mysql_result: dict[str, object] = {"mysql_import_status": "SKIPPED"}
|
if not args.skip_mysql:
|
mysql_result = import_mysql(args, cases, source_hash, source_rel, out_dir)
|
|
body_files = sorted(body_dir.glob("*.md"))
|
summary = {
|
"run_id": "legacy_case_import_20260624",
|
"import_batch_id": IMPORT_BATCH_ID,
|
"source_document_path": source_rel,
|
"source_hash": source_hash,
|
"output_dir": project_rel(out_dir),
|
"case_count": len(cases),
|
"case_body_file_count": len(body_files),
|
"file_export_status": "PASS" if len(cases) == len(body_files) else "FAIL",
|
**mysql_result,
|
}
|
summary_json.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
write_manifest(manifest_csv, [index_csv, summary_json, *body_files])
|
summary["manifest_path"] = project_rel(manifest_csv)
|
summary["manifest_sha256"] = sha256_file(manifest_csv)
|
summary_json.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
write_manifest(manifest_csv, [index_csv, summary_json, *body_files])
|
|
print(json.dumps(summary, ensure_ascii=False, indent=2))
|
|
|
if __name__ == "__main__":
|
main()
|
