mbx/mb-project-info.git

import argparse
import csv
import hashlib
import re
from collections import Counter
from datetime import datetime, timezone, timedelta
from pathlib import Path
 
 
METAL_KEYWORDS = {
    "铜": ["铜", "铜矿", "电解铜", "阴极铜", "沪铜", "伦铜"],
    "铝": ["铝", "氧化铝", "电解铝", "沪铝", "铝土矿"],
    "金": ["黄金", "金价", "贵金属", "COMEX金", "伦敦金"],
    "银": ["白银", "银价", "银浆"],
    "锂": ["锂", "碳酸锂", "氢氧化锂", "锂矿", "盐湖"],
    "钴": ["钴"],
    "镍": ["镍"],
    "稀土": ["稀土", "氧化镨钕", "镨钕", "钕铁硼", "磁材"],
    "锡": ["锡"],
    "钨": ["钨"],
    "钼": ["钼"],
    "锑": ["锑"],
}
 
THEME_KEYWORDS = {
    "price": ["价格", "价差", "涨", "跌", "均价", "报价", "沪", "LME", "COMEX"],
    "inventory": ["库存", "累库", "去库", "仓单"],
    "supply": ["供给", "供应", "产量", "产能", "矿山", "冶炼", "开工", "进口"],
    "demand": ["需求", "消费", "订单", "出货", "下游", "终端"],
    "cost": ["成本", "利润", "毛利", "加工费", "TC", "电价"],
    "policy": ["政策", "关税", "配额", "制裁", "出口管制", "收储"],
    "risk": ["风险", "扰动", "不及预期", "下滑", "压力", "亏损"],
    "company": ["公司", "股份", "集团", "矿业", "锂业", "铝业", "铜业", "黄金", "稀土"],
    "project": ["项目", "扩建", "投产", "建设", "收购", "资源量", "储量"],
}
 
COMPANY_PATTERN = re.compile(
    r"([\u4e00-\u9fa5A-Za-z0-9]{2,18}(?:股份|集团|矿业|锂业|铝业|铜业|黄金|稀土|钴业|能源|材料|资源|科技|新材|金属|冶炼|化工))"
)
STOCK_PATTERN = re.compile(r"(?:股票代码|代码|证券代码)?[:： ]?([036]\d{5})(?:\.[A-Z]{2})?")
 
 
def read_csv(path):
    with path.open("r", encoding="utf-8-sig", newline="") as handle:
        return list(csv.DictReader(handle))
 
 
def write_csv(path, fieldnames, rows):
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("w", encoding="utf-8-sig", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerows(rows)
 
 
def norm_path(value):
    return (value or "").replace("\\", "/").strip()
 
 
def text_hash(value):
    return hashlib.sha256(value.encode("utf-8")).hexdigest()[:16]
 
 
def split_sentences(text):
    text = re.sub(r"[ \t]+", " ", text)
    parts = re.split(r"(?<=[。！？；;])|\n+", text)
    return [part.strip() for part in parts if len(part.strip()) >= 18]
 
 
def scan_text(path):
    current_page = ""
    for line in path.read_text(encoding="utf-8", errors="ignore").splitlines():
        page_match = re.match(r"\[\[PAGE\s+(\d+)\]\]", line.strip())
        if page_match:
            current_page = page_match.group(1)
            continue
        for sentence in split_sentences(line):
            yield current_page, sentence
 
 
def tags_for(sentence, keyword_map):
    tags = []
    for tag, keywords in keyword_map.items():
        if any(keyword in sentence for keyword in keywords):
            tags.append(tag)
    return tags
 
 
def fact_type(sentence, theme_tags):
    if "risk" in theme_tags:
        return "risk_sentence"
    if re.search(r"\d+(?:\.\d+)?\s*(?:%|万吨|吨|亿元|元/吨|美元/吨|万亿元|GWh|万台)", sentence):
        return "indicator_sentence"
    return "fact_or_view_sentence"
 
 
def confidence(sentence, metal_tags, theme_tags):
    if not metal_tags:
        return "LOW"
    if len(theme_tags) >= 2 or re.search(r"\d", sentence):
        return "MEDIUM"
    return "LOW"
 
 
def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--project-root", default=".")
    parser.add_argument("--conversion-status", required=True)
    parser.add_argument("--extracted-dir", required=True)
    parser.add_argument("--manifest-dir", required=True)
    parser.add_argument("--outputs-dir", required=True)
    parser.add_argument("--case-id", default="ANA-YS-INDUSTRY-001")
    parser.add_argument("--batch-id", default="BATCH-003")
    parser.add_argument("--run-id", default="RUN-ANA-YS-INDUSTRY-001-BATCH-003-EXTRACT-001")
    parser.add_argument("--max-facts-per-doc", type=int, default=12)
    args = parser.parse_args()
 
    project_root = Path(args.project_root).resolve()
    status_rows = read_csv(project_root / args.conversion_status)
    extracted_dir = project_root / args.extracted_dir
    manifest_dir = project_root / args.manifest_dir
    outputs_dir = project_root / args.outputs_dir
    created_at = datetime.now(timezone(timedelta(hours=8))).isoformat(timespec="seconds")
 
    fact_rows = []
    class_rows = []
    company_rows = []
 
    converted_rows = [row for row in status_rows if row.get("conversion_status") == "TEXT_CONVERTED"]
    for doc_index, row in enumerate(converted_rows, start=1):
        converted_path = project_root / row.get("converted_text_path", "")
        if not converted_path.exists():
            continue
        doc_id = row.get("source_doc_id") or f"YS-B3-DOC-{doc_index:03d}"
        doc_metal_counter = Counter()
        doc_theme_counter = Counter()
        doc_fact_count = 0
        company_seen = set()
 
        for page, sentence in scan_text(converted_path):
            metal_tags = tags_for(sentence, METAL_KEYWORDS)
            theme_tags = tags_for(sentence, THEME_KEYWORDS)
            if metal_tags:
                doc_metal_counter.update(metal_tags)
            if theme_tags:
                doc_theme_counter.update(theme_tags)
 
            if metal_tags and theme_tags and doc_fact_count < args.max_facts_per_doc:
                doc_fact_count += 1
                fact_rows.append(
                    {
                        "fact_id": f"YS-B3-F-{len(fact_rows) + 1:05d}",
                        "case_id": args.case_id,
                        "batch_id": args.batch_id,
                        "sub_batch_id": row.get("sub_batch_id", ""),
                        "run_id": args.run_id,
                        "doc_id": doc_id,
                        "fact_type": fact_type(sentence, theme_tags),
                        "metal_tags": "|".join(metal_tags),
                        "theme_tags": "|".join(theme_tags),
                        "evidence_text": sentence[:220],
                        "source_location": f"converted_text_page_{page or 'unknown'}_sentence_scan",
                        "page_no": page,
                        "converted_text_path": row.get("converted_text_path", ""),
                        "raw_file_path": row.get("raw_file_path", ""),
                        "raw_sha256": row.get("raw_sha256", ""),
                        "confidence": confidence(sentence, metal_tags, theme_tags),
                        "review_status": "DRAFT_FOR_REVIEW",
                        "created_at": created_at,
                    }
                )
 
            for company_match in COMPANY_PATTERN.findall(sentence):
                if company_match in company_seen:
                    continue
                company_seen.add(company_match)
                stock_match = STOCK_PATTERN.search(sentence)
                company_rows.append(
                    {
                        "company_map_id": f"YS-B3-CM-{len(company_rows) + 1:05d}",
                        "case_id": args.case_id,
                        "batch_id": args.batch_id,
                        "sub_batch_id": row.get("sub_batch_id", ""),
                        "run_id": args.run_id,
                        "doc_id": doc_id,
                        "company_name": company_match,
                        "stock_code": stock_match.group(1) if stock_match else "",
                        "metal_tags": "|".join(tags_for(sentence, METAL_KEYWORDS)),
                        "context_snippet": sentence[:220],
                        "source_location": f"converted_text_page_{page or 'unknown'}_company_scan",
                        "page_no": page,
                        "converted_text_path": row.get("converted_text_path", ""),
                        "raw_file_path": row.get("raw_file_path", ""),
                        "confidence": "LOW",
                        "review_status": "DRAFT_FOR_REVIEW",
                        "created_at": created_at,
                    }
                )
 
        class_rows.append(
            {
                "case_id": args.case_id,
                "batch_id": args.batch_id,
                "sub_batch_id": row.get("sub_batch_id", ""),
                "run_id": args.run_id,
                "doc_id": doc_id,
                "raw_file_path": row.get("raw_file_path", ""),
                "converted_text_path": row.get("converted_text_path", ""),
                "top_metal_tags": "|".join([tag for tag, _ in doc_metal_counter.most_common(6)]),
                "top_theme_tags": "|".join([tag for tag, _ in doc_theme_counter.most_common(8)]),
                "fact_count": doc_fact_count,
                "company_mention_count": len(company_seen),
                "review_status": "DRAFT_FOR_REVIEW",
                "created_at": created_at,
            }
        )
 
    fact_path = extracted_dir / "evidence_fact_table_batch003_extract001.csv"
    class_path = extracted_dir / "classification_summary_batch003_extract001.csv"
    company_path = extracted_dir / "company_exposure_batch003_extract001.csv"
 
    write_csv(
        fact_path,
        [
            "fact_id",
            "case_id",
            "batch_id",
            "sub_batch_id",
            "run_id",
            "doc_id",
            "fact_type",
            "metal_tags",
            "theme_tags",
            "evidence_text",
            "source_location",
            "page_no",
            "converted_text_path",
            "raw_file_path",
            "raw_sha256",
            "confidence",
            "review_status",
            "created_at",
        ],
        fact_rows,
    )
    write_csv(
        class_path,
        [
            "case_id",
            "batch_id",
            "sub_batch_id",
            "run_id",
            "doc_id",
            "raw_file_path",
            "converted_text_path",
            "top_metal_tags",
            "top_theme_tags",
            "fact_count",
            "company_mention_count",
            "review_status",
            "created_at",
        ],
        class_rows,
    )
    write_csv(
        company_path,
        [
            "company_map_id",
            "case_id",
            "batch_id",
            "sub_batch_id",
            "run_id",
            "doc_id",
            "company_name",
            "stock_code",
            "metal_tags",
            "context_snippet",
            "source_location",
            "page_no",
            "converted_text_path",
            "raw_file_path",
            "confidence",
            "review_status",
            "created_at",
        ],
        company_rows,
    )
 
    manifest_rows = []
    for artifact_path, artifact_type, row_count in [
        (fact_path, "evidence_fact_table", len(fact_rows)),
        (class_path, "classification_summary", len(class_rows)),
        (company_path, "company_exposure", len(company_rows)),
    ]:
        data = artifact_path.read_bytes()
        manifest_rows.append(
            {
                "case_id": args.case_id,
                "batch_id": args.batch_id,
                "run_id": args.run_id,
                "artifact_type": artifact_type,
                "artifact_path": artifact_path.relative_to(project_root).as_posix(),
                "row_count": row_count,
                "sha256": hashlib.sha256(data).hexdigest(),
                "review_status": "DRAFT_FOR_REVIEW",
                "created_at": created_at,
            }
        )
    manifest_path = manifest_dir / "batch003_extract001_manifest.csv"
    write_csv(
        manifest_path,
        ["case_id", "batch_id", "run_id", "artifact_type", "artifact_path", "row_count", "sha256", "review_status", "created_at"],
        manifest_rows,
    )
 
    by_sub = Counter(row.get("sub_batch_id", "") for row in fact_rows)
    md_lines = [
        "# ANA-YS-INDUSTRY-001 BATCH-003 EXTRACT-001 批次摘要",
        "",
        "状态：DRAFT_FOR_REVIEW",
        f"生成时间：{created_at}",
        f"case_id：{args.case_id}",
        f"batch_id：{args.batch_id}",
        f"run_id：{args.run_id}",
        "",
        "## 本轮动作",
        "",
        "对 BATCH-003 已转换文本执行规则扫描，生成事实句、分类汇总和公司映射草表。事实句保留 converted text 页码级定位；该定位仍需人工复核，不等于正式引用页码。",
        "",
        "## 输出",
        "",
        "| 文件 | 行数 | 用途 |",
        "|---|---:|---|",
        f"| `{fact_path.relative_to(project_root).as_posix()}` | {len(fact_rows)} | 事实句、指标句、风险句草表 |",
        f"| `{class_path.relative_to(project_root).as_posix()}` | {len(class_rows)} | 每篇研报金属和主题标签 |",
        f"| `{company_path.relative_to(project_root).as_posix()}` | {len(company_rows)} | 公司映射和上下文片段草表 |",
        f"| `{manifest_path.relative_to(project_root).as_posix()}` | {len(manifest_rows)} | 本轮产物清单和 hash |",
        "",
        "## sub_batch 事实句数量",
        "",
        "| sub_batch_id | 事实句 |",
        "|---|---:|",
    ]
    for sub_batch_id in sorted(by_sub):
        md_lines.append(f"| {sub_batch_id} | {by_sub[sub_batch_id]} |")
    md_lines.extend(
        [
            "",
            "## 边界",
            "",
            "本轮仍是规则扫描。`source_location` 为 converted text 页码和句子扫描定位，后续必须补段落、表格定位和人工校验后，才能升级为正式证据。",
        ]
    )
    summary_path = outputs_dir / "batch003_extract001_summary.md"
    summary_path.write_text("\n".join(md_lines) + "\n", encoding="utf-8")
 
    print(f"facts={len(fact_rows)}")
    print(f"classification={len(class_rows)}")
    print(f"companies={len(company_rows)}")
    print(f"manifest={manifest_path.relative_to(project_root).as_posix()}")
    print(f"summary={summary_path.relative_to(project_root).as_posix()}")
 
 
if __name__ == "__main__":
    main()