importer/mapper.py

from __future__ import annotations

import json
import re
from hashlib import sha256
from datetime import datetime, timezone
from typing import Any


_URL_RE = re.compile(r"https?://\S+")
_NOTE_ID_RE = re.compile(r"/explore/([^/?#]+)")
_USER_ID_RE = re.compile(r"/user/profile/([^/?#]+)")


def clean_cell(value: Any) -> Any:
    if value is None:
        return None
    if isinstance(value, str):
        s = value.strip()
        return s if s else None
    return value


def clean_raw_row(raw: dict[str, Any]) -> dict[str, Any]:
    cleaned: dict[str, Any] = {}
    for k, v in raw.items():
        key = str(k).strip() if k is not None else ""
        if not key:
            continue
        cleaned[key] = clean_cell(v)
    return cleaned


def _pick_by_headers(raw: dict[str, Any], candidates: set[str]) -> Any:
    for k, v in raw.items():
        if str(k).strip().lower() in candidates and v is not None:
            return v
    return None


def _find_first_url(raw: dict[str, Any]) -> str | None:
    for v in raw.values():
        if isinstance(v, str):
            m = _URL_RE.search(v)
            if m:
                return m.group(0)
    return None


def _extract_note_id(url: str) -> str | None:
    m = _NOTE_ID_RE.search(url)
    return m.group(1) if m else None


def _extract_user_id(url: str) -> str | None:
    m = _USER_ID_RE.search(url)
    return m.group(1) if m else None


def _parse_int(value: Any) -> int | None:
    if value is None:
        return None
    if isinstance(value, bool):
        return int(value)
    if isinstance(value, int):
        return value
    if isinstance(value, float):
        return int(value)
    if isinstance(value, str):
        s = value.strip()
        if not s:
            return None
        s = s.replace(",", "").replace("，", "")
        if s.endswith("万"):
            try:
                return int(float(s[:-1]) * 10000)
            except Exception:
                return None
        if s.endswith("w"):
            try:
                return int(float(s[:-1]) * 10000)
            except Exception:
                return None
        m = re.search(r"-?\d+(\.\d+)?", s)
        if not m:
            return None
        try:
            return int(float(m.group(0)))
        except Exception:
            return None
    return None


def _parse_time_iso(value: Any) -> str | None:
    if value is None:
        return None
    if isinstance(value, datetime):
        return value.isoformat()
    if isinstance(value, (int, float)) and not isinstance(value, bool):
        ts = float(value)
        try:
            if ts > 10_000_000_000:
                ts = ts / 1000.0
            return datetime.fromtimestamp(ts, tz=timezone.utc).isoformat()
        except Exception:
            return None
    if isinstance(value, str):
        s = value.strip()
        if not s:
            return None
        try:
            return datetime.fromisoformat(s).isoformat()
        except Exception:
            pass
        for fmt in (
            "%Y-%m-%d %H:%M:%S",
            "%Y/%m/%d %H:%M:%S",
            "%Y-%m-%d %H:%M",
            "%Y/%m/%d %H:%M",
            "%Y-%m-%d",
            "%Y/%m/%d",
            "%Y.%m.%d %H:%M:%S",
            "%Y.%m.%d %H:%M",
            "%Y.%m.%d",
        ):
            try:
                return datetime.strptime(s, fmt).isoformat()
            except Exception:
                continue
    return None


def compute_row_hash(raw: dict[str, Any]) -> str:
    data = json.dumps(raw, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8")
    return sha256(data).hexdigest()


def map_row(raw_row: dict[str, Any]) -> tuple[dict[str, Any] | None, str | None, list[str]]:
    issues: list[str] = []
    raw = clean_raw_row(raw_row)

    url = _pick_by_headers(
        raw,
        {
            "url",
            "link",
            "note_url",
            "user_url",
            "note link",
            "链接",
            "笔记url",
            "笔记链接",
            "笔记地址",
            "作品链接",
            "作品url",
            "作品地址",
            "主页链接",
            "用户主页",
            "用户url",
        },
    )
    if isinstance(url, str):
        url = url.strip()
    else:
        url = None

    if not url:
        url = _find_first_url(raw)

    query = _pick_by_headers(raw, {"query", "keyword", "关键词", "搜索词"})
    if isinstance(query, str):
        query = query.strip()
    else:
        query = None

    title = _pick_by_headers(raw, {"title", "标题", "笔记标题", "作品标题"})
    if isinstance(title, str):
        title = title.strip()
    else:
        title = None

    nickname = _pick_by_headers(raw, {"nickname", "author", "作者", "昵称", "作者昵称", "博主昵称", "博主"})
    if isinstance(nickname, str):
        nickname = nickname.strip()
    else:
        nickname = None

    publish_time = _pick_by_headers(
        raw,
        {
            "publish_time",
            "published_at",
            "发布时间",
            "发布/发布时间",
            "发布",
            "发布时间(北京时间)",
            "发布时间（北京时间）",
            "上传时间",
        },
    )
    publish_time_iso = _parse_time_iso(publish_time)

    like_count = _parse_int(
        _pick_by_headers(raw, {"like", "likes", "点赞", "点赞数", "赞", "赞数", "获赞", "喜欢"}),
    )
    comment_count = _parse_int(_pick_by_headers(raw, {"comment", "comments", "评论", "评论数"}))
    collect_count = _parse_int(_pick_by_headers(raw, {"collect", "collects", "收藏", "收藏数"}))
    share_count = _parse_int(_pick_by_headers(raw, {"share", "shares", "分享", "分享数", "转发", "转发数"}))

    exposure = _parse_int(_pick_by_headers(raw, {"exposure", "曝光", "曝光量"}))
    read_count = _parse_int(_pick_by_headers(raw, {"read", "阅读", "阅读量"}))
    view_count = _parse_int(_pick_by_headers(raw, {"view", "views", "浏览", "浏览量"}))
    interact_count = _parse_int(_pick_by_headers(raw, {"interact", "interaction", "互动", "互动量"}))
    follow_count = _parse_int(_pick_by_headers(raw, {"follow", "转粉", "涨粉"}))

    content_type = _pick_by_headers(raw, {"content_type", "内容类型", "笔记类型", "内容形式"})
    if isinstance(content_type, str):
        content_type = content_type.strip()
    else:
        content_type = None

    normalized: dict[str, Any] = {}
    dedup_key: str | None = None

    if url and "xiaohongshu.com" in url:
        note_id = _extract_note_id(url)
        user_id = _extract_user_id(url)
        if note_id:
            normalized["kind"] = "note"
            normalized["note_url"] = url
            normalized["note_id"] = note_id
            dedup_key = f"note:{note_id}"
        elif user_id:
            normalized["kind"] = "user"
            normalized["user_url"] = url
            normalized["user_id"] = user_id
            dedup_key = f"user:{user_id}"
        else:
            normalized["kind"] = "url"
            normalized["url"] = url
            dedup_key = f"url:{url}"
    elif query:
        normalized["kind"] = "search"
        normalized["query"] = query
        dedup_key = f"search:{query}"
    elif url:
        normalized["kind"] = "url"
        normalized["url"] = url
        dedup_key = f"url:{url}"
    else:
        issues.append("未识别到 url 或 query")

    if title:
        normalized["title"] = title
    if nickname:
        normalized["nickname"] = nickname
        normalized["author"] = nickname

    if publish_time_iso:
        normalized["publish_time"] = publish_time_iso
    if like_count is not None:
        normalized["like_count"] = like_count
    if comment_count is not None:
        normalized["comment_count"] = comment_count
    if collect_count is not None:
        normalized["collect_count"] = collect_count
    if share_count is not None:
        normalized["share_count"] = share_count
    if exposure is not None:
        normalized["exposure"] = exposure
    if read_count is not None:
        normalized["read"] = read_count
    if view_count is not None:
        normalized["view"] = view_count
    if interact_count is not None:
        normalized["interact"] = interact_count
    if follow_count is not None:
        normalized["follow"] = follow_count
    if content_type:
        normalized["content_type"] = content_type
    if not normalized:
        return None, None, issues

    return normalized, dedup_key, issues