"""Local announcement and annual-report text research.

The module deliberately has no network or NLP-model dependency.  It provides a
small, deterministic text signal layer that can be used by later stages of the
pipeline and is explicit about keyword negation.  Scores are research signals,
not a substitute for reading the source document.
"""

from __future__ import annotations

from dataclasses import dataclass
import json
from pathlib import Path
import re
from typing import Any, Iterable, Mapping

import numpy as np
import pandas as pd


TEXT_RESEARCH_MODEL_VERSION = "0.1.0"
SUPPORTED_SUFFIXES = frozenset({".txt", ".md", ".json", ".csv"})

# Keep this list stable: callers can safely concatenate the result with later
# feature tables even when no local documents have been supplied.
TEXT_RESEARCH_COLUMNS = [
    "document_id", "source_path", "file_name", "file_type", "record_index",
    "source_code", "security_id", "title", "announcement_date", "report_period",
    "announcement_type", "text_length", "text_preview",
    "keyword_hits", "negated_keyword_hits", "negated_terms", "risk_keyword_hits",
    "catalyst_keyword_hits", "moat_keyword_hits", "management_keyword_hits",
    "risk_terms", "catalyst_terms", "moat_terms", "management_terms",
    "risk_score", "catalyst_score", "moat_score", "management_score",
    "management_risk_score", "risk_level", "catalyst_signal",
    "moat_assessment", "management_assessment", "evidence", "model_version",
]


# Values are intentionally short Chinese phrases so this remains useful on
# annual reports and exchange announcements without requiring tokenization.
_LEXICONS: dict[str, dict[str, float]] = {
    "risk": {
        "重大风险": 2.0, "经营风险": 1.5, "诉讼": 1.5, "仲裁": 1.2,
        "处罚": 1.8, "违规": 1.8, "减值": 1.4, "商誉": 1.2, "流动性风险": 1.8,
        "债务": 1.0, "违约": 2.0, "亏损": 1.3, "下滑": 0.8, "质押": 1.5,
        "冻结": 1.6, "退市": 2.0, "审计意见": 1.8, "内控缺陷": 1.8,
        "监管": 1.2, "调查": 1.5, "风险": 0.7,
    },
    "catalyst": {
        "订单": 1.0, "中标": 1.5, "扩产": 1.2, "投产": 1.3, "提价": 1.2,
        "涨价": 1.2, "回购": 1.2, "分红": 0.8, "增持": 1.3, "并购": 1.0,
        "重组": 1.0, "获批": 1.2, "突破": 1.1, "创新": 0.8, "战略合作": 1.3,
        "产能": 0.8, "业绩预增": 1.8, "增长": 0.7, "签约": 1.1, "新产品": 1.0,
    },
    "moat": {
        "专利": 1.2, "技术壁垒": 2.0, "护城河": 2.0, "品牌": 1.0, "客户粘性": 1.8,
        "转换成本": 1.8, "规模效应": 1.5, "网络效应": 1.5, "成本优势": 1.4,
        "牌照": 1.2, "认证": 0.9, "市场份额": 1.0, "独家": 1.3, "核心技术": 1.7,
        "研发": 0.8, "长期合作": 1.1,
    },
    "management": {
        "管理层": 0.8, "董事会": 0.8, "高管": 0.8, "股权激励": 1.8, "员工持股": 1.4,
        "回购": 0.8, "分红": 0.7, "独立董事": 0.8, "审计委员会": 1.0, "薪酬": 0.6,
        "诚信": 1.0, "内部控制": 1.0, "内控": 0.8, "实际控制人": 0.7,
        "关联交易": -1.2, "违规": -2.0, "处罚": -2.0, "调查": -1.5,
        "失信": -2.0, "侵占": -2.0, "财务造假": -3.0,
    },
}

_NEGATION_WORDS = (
    "不构成", "不存在", "未发现", "未见", "尚未", "尚无", "没有", "无法",
    "难以", "否认", "未", "无", "不", "no", "not", "without",
)
_NEGATION_RE = re.compile("|".join(sorted(map(re.escape, _NEGATION_WORDS), key=len, reverse=True)), re.I)
_DATE_RE = re.compile(r"(?<!\d)(20\d{2})[-/.年](\d{1,2})[-/.月](\d{1,2})日?(?!\d)")
# Do not match the ``2025年`` prefix of a publication date.  A report period
# is inferred from an explicit report/annual-report label only.
_YEAR_RE = re.compile(r"(?<!\d)(20\d{2})(?:年年度报告|年度报告|年报|财报)(?!\d)")
_CODE_RE = re.compile(r"(?i)(?<![a-z0-9])(?:sh|sz|bj)[ ._-]?(\d{6})(?![a-z0-9])")
_BARE_CODE_RE = re.compile(r"(?<!\d)(\d{6})(?!\d)")


@dataclass(frozen=True)
class KeywordHit:
    term: str
    dimension: str
    weight: float
    negated: bool
    sentence: str


def empty_text_research() -> pd.DataFrame:
    """Return an empty frame with the stable text-research schema."""
    return pd.DataFrame(columns=TEXT_RESEARCH_COLUMNS)


def discover_text_files(config_dir: str | Path | None) -> list[Path]:
    """Recursively discover supported local text files in deterministic order."""
    if config_dir is None:
        return []
    root = Path(config_dir)
    if not root.exists() or not root.is_dir():
        return []
    return sorted(
        (path for path in root.rglob("*") if path.is_file() and path.suffix.lower() in SUPPORTED_SUFFIXES),
        key=lambda path: path.as_posix().lower(),
    )


def _decode(path: Path) -> str:
    raw = path.read_bytes()
    for encoding in ("utf-8-sig", "utf-8", "gb18030", "utf-16"):
        try:
            return raw.decode(encoding)
        except UnicodeDecodeError:
            continue
    return raw.decode("utf-8", errors="replace")


def _key_value(mapping: Mapping[str, Any], names: Iterable[str]) -> Any:
    folded = {str(key).strip().lower(): value for key, value in mapping.items()}
    for name in names:
        value = folded.get(name.lower())
        if value is not None and str(value).strip() not in {"", "nan", "none"}:
            return value
    return None


def _as_text(value: Any) -> str:
    if value is None or (isinstance(value, float) and np.isnan(value)):
        return ""
    if isinstance(value, (dict, list, tuple)):
        return json.dumps(value, ensure_ascii=False, default=str)
    return str(value)


def _json_records(value: Any) -> list[dict[str, Any]]:
    if isinstance(value, list):
        records: list[dict[str, Any]] = []
        for item in value:
            records.extend(_json_records(item))
        return records or [{"content": ""}]
    if isinstance(value, dict):
        # A common API shape is {data: [{...}, ...]}.
        for key in ("data", "items", "records", "results", "公告", "announcements"):
            nested = value.get(key)
            if isinstance(nested, list):
                records = _json_records(nested)
                shared = {k: v for k, v in value.items() if k != key and not isinstance(v, (dict, list))}
                return [{**shared, **record} for record in records]
        return [value]
    return [{"content": _as_text(value)}]


def parse_text_file(path: str | Path) -> list[dict[str, Any]]:
    """Read one local source and return one or more normalized raw records.

    CSV rows and JSON list members become separate records; plain text files
    produce one record.  Parsing errors are converted to an empty record so a
    single malformed source cannot break a batch run.
    """
    path = Path(path)
    suffix = path.suffix.lower()
    try:
        if suffix in {".txt", ".md"}:
            return [{"content": _decode(path)}]
        if suffix == ".json":
            return _json_records(json.loads(_decode(path)))
        if suffix == ".csv":
            for encoding in ("utf-8-sig", "utf-8", "gb18030"):
                try:
                    frame = pd.read_csv(path, dtype=str, encoding=encoding).fillna("")
                    break
                except (UnicodeDecodeError, pd.errors.ParserError):
                    frame = None
            if frame is None or frame.empty:
                return []
            return [row.to_dict() for _, row in frame.iterrows()]
    except (OSError, ValueError, json.JSONDecodeError, TypeError):
        return [{"content": ""}]
    return []


def _record_text(record: Mapping[str, Any]) -> str:
    content_names = ("content", "text", "body", "正文", "全文", "内容", "公告内容", "document")
    content = _key_value(record, content_names)
    if content is not None:
        # Include title and metadata in the text only when they are not already
        # part of the content, so title-only announcements still score.
        parts = [_as_text(content)]
        for key in ("title", "公告标题", "announcement_title", "summary", "摘要"):
            value = _key_value(record, (key,))
            if value is not None and _as_text(value) not in parts:
                parts.append(_as_text(value))
        return "\n".join(part for part in parts if part)
    values = [_as_text(value) for value in record.values() if _as_text(value)]
    return "\n".join(values)


def _date_value(value: Any, text: str) -> str | None:
    raw = _as_text(value)
    labelled = re.search(r"(?:公告日期|发布日期|披露日期|发布时间|发布日)\s*[:：]\s*(20\d{2}[-/.年]\d{1,2}[-/.月]\d{1,2}日?)", text)
    match = (
        _DATE_RE.search(raw)
        or (_DATE_RE.search(labelled.group(1)) if labelled else None)
        or _DATE_RE.search(text)
    )
    if match:
        year, month, day = map(int, match.groups())
        try:
            return f"{year:04d}-{month:02d}-{day:02d}"
        except ValueError:
            return None
    return None


def _report_period(value: Any, text: str) -> str | None:
    raw = _as_text(value)
    period_text = raw or text
    half_match = re.search(r"(?<!\d)(20\d{2})年(?:半年度报告|半年报|中报)", period_text)
    if half_match:
        return f"{half_match.group(1)}-06-30"
    quarter_match = re.search(r"(?<!\d)(20\d{2})年(?:第?([一二三四1234])季度报告|Q([1-4]))", period_text, re.I)
    if quarter_match:
        quarter = quarter_match.group(2) or quarter_match.group(3)
        if quarter in {"一", "二", "三", "四"}:
            quarter_number = {"一": 1, "二": 2, "三": 3, "四": 4}[quarter]
        else:
            quarter_number = int(quarter) if quarter and quarter.isdigit() else None
        if quarter_number:
            return f"{quarter_match.group(1)}-{quarter_number * 3:02d}-{[31, 30, 30, 31][quarter_number - 1]:02d}"
    # Annual/period labels are more authoritative than a nearby publication
    # date (for example, ``2024年年度报告\n公告日期：2025年3月``).
    year_match = _YEAR_RE.search(raw) or _YEAR_RE.search(text)
    if year_match:
        return f"{year_match.group(1)}-12-31"
    match = _DATE_RE.search(raw) or _DATE_RE.search(text)
    if match:
        year, month, day = map(int, match.groups())
        try:
            return f"{year:04d}-{month:02d}-{day:02d}"
        except ValueError:
            pass
    return None


def _security_code(text: str) -> tuple[str | None, str | None]:
    match = _CODE_RE.search(text)
    if match:
        prefix_match = re.search(r"(?i)(sh|sz|bj)", match.group(0))
        prefix = prefix_match.group(1).upper() if prefix_match else None
        code = match.group(1)
        return f"{prefix}.{code}" if prefix else None, f"{prefix.lower()}{code}" if prefix else None
    # Avoid treating years and dates as stock codes.  Common A-share prefixes
    # make this conservative enough for unlabelled CSV exports.
    for match in _BARE_CODE_RE.finditer(text):
        code = match.group(1)
        if code.startswith(("000", "001", "002", "003", "300", "301", "600", "601", "603", "605", "688", "689", "830", "831", "832", "833", "834", "835", "836", "837", "838", "839", "870", "871", "872")):
            prefix = "SH" if code.startswith(("600", "601", "603", "605", "688", "689")) else ("BJ" if code.startswith(("8",)) else "SZ")
            return f"{prefix}.{code}", f"{prefix.lower()}{code}"
    return None, None


def _sentences(text: str) -> list[str]:
    return [sentence.strip() for sentence in re.split(r"(?<=[。！？!?;；\n])", text) if sentence.strip()]


def _is_negated(sentence: str, start: int) -> bool:
    # Negation is scoped to the same sentence and a short preceding window.
    # This avoids making "风险已解除" positive merely because a negation
    # appears in an unrelated paragraph.
    prefix = sentence[:start]
    # A comma generally starts a new clause in Chinese announcements.  Do not
    # let "不存在重大风险，核心技术..." negate the later, unrelated terms.
    prefix = re.split(r"[,，、:：]", prefix)[-1]
    prefix = prefix[-18:]
    return bool(_NEGATION_RE.search(prefix))


def _collect_hits(text: str) -> list[KeywordHit]:
    hits: list[KeywordHit] = []
    for sentence in _sentences(text):
        for dimension, lexicon in _LEXICONS.items():
            candidates: list[tuple[int, int, str, float, bool]] = []
            for term, weight in lexicon.items():
                for match in re.finditer(re.escape(term), sentence, flags=re.I):
                    candidates.append((match.start(), match.end(), term, weight, _is_negated(sentence, match.start())))
            # Prefer the longest term when lexicon entries overlap (for
            # example, ``重大风险`` contains the generic ``风险`` keyword).
            # Otherwise one phrase would be counted twice and distort both
            # hit counts and the negation scope.
            selected: list[tuple[int, int, str, float, bool]] = []
            for candidate in sorted(candidates, key=lambda item: (item[0], -(item[1] - item[0]))):
                if any(candidate[0] < prior[1] and prior[0] < candidate[1] for prior in selected):
                    continue
                selected.append(candidate)
            hits.extend(KeywordHit(term, dimension, weight, negated, sentence[:160]) for _, _, term, weight, negated in selected)
    return hits


def _dimension_score(hits: list[KeywordHit], dimension: str, *, positive_only: bool = False) -> float:
    values = [hit for hit in hits if hit.dimension == dimension and not hit.negated]
    if positive_only:
        values = [hit for hit in values if hit.weight > 0]
    raw = sum(hit.weight for hit in values)
    if raw <= 0:
        return 0.0
    return round(float(min(100.0, 100.0 * raw / (raw + 8.0))), 2)


def score_text(text: str) -> dict[str, Any]:
    """Score one document and return hit counts, scores and short evidence."""
    text = _as_text(text)
    hits = _collect_hits(text)
    active = [hit for hit in hits if not hit.negated]
    by_dimension: dict[str, list[KeywordHit]] = {dimension: [h for h in active if h.dimension == dimension] for dimension in _LEXICONS}
    unique = lambda values: ";".join(dict.fromkeys(hit.term for hit in values))
    negated_terms = ";".join(dict.fromkeys(hit.term for hit in hits if hit.negated))
    evidence = []
    for dimension, values in by_dimension.items():
        if values:
            evidence.append(f"{dimension}:" + "；".join(f"{hit.term}({hit.sentence[:60]})" for hit in values[:4]))
    risk_score = _dimension_score(hits, "risk")
    catalyst_score = _dimension_score(hits, "catalyst", positive_only=True)
    moat_score = _dimension_score(hits, "moat", positive_only=True)
    management_quality = _dimension_score(hits, "management", positive_only=True)
    management_risk_raw = sum(abs(hit.weight) for hit in hits if hit.dimension == "management" and not hit.negated and hit.weight < 0)
    management_risk_score = round(float(min(100.0, 100.0 * management_risk_raw / (management_risk_raw + 8.0))), 2) if management_risk_raw else 0.0
    return {
        "keyword_hits": len(active), "negated_keyword_hits": len(hits) - len(active), "negated_terms": negated_terms,
        "risk_keyword_hits": len(by_dimension["risk"]), "catalyst_keyword_hits": len(by_dimension["catalyst"]),
        "moat_keyword_hits": len(by_dimension["moat"]), "management_keyword_hits": len(by_dimension["management"]),
        "risk_terms": unique(by_dimension["risk"]), "catalyst_terms": unique(by_dimension["catalyst"]),
        "moat_terms": unique(by_dimension["moat"]), "management_terms": unique(by_dimension["management"]),
        "risk_score": risk_score, "catalyst_score": catalyst_score, "moat_score": moat_score,
        "management_score": management_quality, "management_risk_score": management_risk_score,
        "risk_level": "高" if risk_score >= 55 else ("中" if risk_score >= 25 else "低"),
        "catalyst_signal": "强" if catalyst_score >= 55 else ("中" if catalyst_score >= 25 else "弱"),
        "moat_assessment": "强" if moat_score >= 55 else ("中" if moat_score >= 25 else "弱"),
        "management_assessment": "偏弱" if management_risk_score >= 35 and management_quality < 45 else ("较好" if management_quality >= 35 else "待核实"),
        "evidence": " | ".join(evidence),
    }


def _announcement_type(title: str, text: str) -> str:
    value = f"{title}\n{text[:400]}"
    choices = (
        ("年度报告", "annual_report"), ("年报", "annual_report"), ("半年度报告", "semiannual_report"),
        ("季度报告", "quarterly_report"), ("业绩预告", "earnings_forecast"), ("业绩快报", "earnings_flash"),
        ("风险提示", "risk_notice"), ("股份回购", "buyback"), ("回购", "buyback"),
        ("股东增持", "shareholder_buy"), ("增持", "shareholder_buy"), ("重大合同", "major_contract"),
        ("中标", "bid_win"), ("资产重组", "restructuring"), ("并购", "merger_acquisition"),
    )
    for marker, result in choices:
        if marker in value:
            return result
    return "other"


def _normalize_record(path: Path, record: Mapping[str, Any], index: int) -> dict[str, Any]:
    text = _record_text(record)
    title = _key_value(record, ("title", "公告标题", "announcement_title", "subject", "标题"))
    title_text = _as_text(title).strip() if title is not None else ""
    if not title_text:
        labelled_title = re.search(r"(?m)^\s*(?:公告标题|公告名称|标题|证券简称)\s*[:：]\s*(.+?)\s*$", text)
        if labelled_title:
            title_text = labelled_title.group(1).strip()
    if not title_text:
        heading = re.search(r"(?m)^\s*#+\s*(.+?)\s*$", text)
        title_text = heading.group(1).strip() if heading else (next((line.strip() for line in text.splitlines() if line.strip()), path.stem))
    source_code_id, source_code = _security_code(path.name + "\n" + " ".join([_as_text(value) for value in record.values()]) + "\n" + text)
    announcement_date = _date_value(_key_value(record, ("announcement_date", "date", "发布日期", "公告日期", "发布日")), text)
    report_period = _report_period(_key_value(record, ("report_period", "period", "报告期", "报告期间")), title_text + "\n" + text)
    scores = score_text(text)
    relative = path.resolve()
    return {
        "document_id": f"{relative.as_posix()}#{index}", "source_path": str(relative), "file_name": path.name,
        "file_type": path.suffix.lower().lstrip("."), "record_index": index,
        "source_code": source_code, "security_id": source_code_id, "title": title_text,
        "announcement_date": announcement_date, "report_period": report_period,
        "announcement_type": _announcement_type(title_text, text), "text_length": len(text),
        "text_preview": re.sub(r"\s+", " ", text).strip()[:300], **scores,
        "model_version": TEXT_RESEARCH_MODEL_VERSION,
    }


def analyze_text_sources(config_dir: str | Path | None, *, max_chars: int = 2_000_000) -> pd.DataFrame:
    """Discover and analyze all supported local files below ``config_dir``."""
    rows: list[dict[str, Any]] = []
    for path in discover_text_files(config_dir):
        for index, record in enumerate(parse_text_file(path)):
            text = _record_text(record)
            if max_chars > 0 and len(text) > max_chars:
                record = dict(record)
                content_key = next((key for key in record if str(key).lower() in {"content", "text", "body", "正文", "全文", "内容", "公告内容", "document"}), None)
                if content_key is not None:
                    record[content_key] = _as_text(record[content_key])[:max_chars]
            rows.append(_normalize_record(path, record, index))
    if not rows:
        return empty_text_research()
    return pd.DataFrame(rows).reindex(columns=TEXT_RESEARCH_COLUMNS)


def build_text_research(settings_or_dir: Any, *, max_chars: int = 2_000_000) -> pd.DataFrame:
    """Convenience alias accepting either a directory or an existing Settings."""
    if hasattr(settings_or_dir, "project_root"):
        root = Path(settings_or_dir.project_root) / "config"
    elif settings_or_dir is not None:
        root = Path(settings_or_dir)
    else:
        root = None
    return analyze_text_sources(root, max_chars=max_chars)


# Descriptive aliases used by orchestration code.
analyze_text_documents = analyze_text_sources
score_text_document = score_text