comment_store.py

"""评论记录的锁内修改；路由负责认证、内容存在性及响应契约。"""
import copy
import asyncio
import logging
import threading
from owned_thread import run_owned_thread
from fastapi import HTTPException
import 数据库连接 as db
from resource_deletion import resource_mutation_lock, assert_comment_target_active
from resource_store import ResourceStoreError


COMMENT_MAX_LENGTH = 5000
MAX_COMMENT_READS = 8
_comment_read_capacity = threading.BoundedSemaphore(MAX_COMMENT_READS)
_comment_read_gate = threading.Lock()
_comment_read_accepting = True
_comment_read_pending = 0
_comment_read_closers = 0
_comment_read_drained = threading.Event()
_comment_read_drained.set()


def start_comment_reads():
    global _comment_read_accepting
    with _comment_read_gate:
        if _comment_read_pending or _comment_read_closers:
            raise RuntimeError("comment reads are still pending")
        _comment_read_accepting = True


def stop_comment_reads():
    global _comment_read_accepting
    with _comment_read_gate:
        _comment_read_accepting = False


async def close_comment_reads():
    global _comment_read_accepting, _comment_read_closers
    with _comment_read_gate:
        _comment_read_accepting = False
        _comment_read_closers += 1
    try:
        cancelled = False
        while not _comment_read_drained.is_set():
            try:
                await asyncio.sleep(.01)
            except asyncio.CancelledError:
                cancelled = True
        if cancelled:
            raise asyncio.CancelledError
    finally:
        with _comment_read_gate:
            _comment_read_closers -= 1


def _item_comments(data, item_id):
    if not isinstance(data, dict):
        raise HTTPException(status_code=503, detail="评论数据暂不可用")
    records = data.get(item_id, [])
    if not isinstance(records, list):
        raise HTTPException(status_code=503, detail="评论数据暂不可用")
    return records


def append_comment(item_id, record, parent_id=None, prepend=False):
    def updater(data):
        records = _item_comments(data, item_id)
        if parent_id:
            parent = next((entry for entry in records if isinstance(entry, dict)
                           and entry.get("id") == parent_id), None)
            if parent is None:
                raise HTTPException(status_code=404, detail="找不到要回复的评论")
            replies = parent.setdefault("replies", [])
            if not isinstance(replies, list):
                raise HTTPException(status_code=503, detail="评论数据暂不可用")
            replies.append(copy.deepcopy(record))
        elif prepend:
            records.insert(0, copy.deepcopy(record))
        else:
            records.append(copy.deepcopy(record))
        data[item_id] = records
    try:
        with resource_mutation_lock:
            assert_comment_target_active(db, item_id)
            db.atomic_update("comments.json", updater, default_data={})
    except ResourceStoreError as error:
        raise HTTPException(status_code=error.status_code, detail=error.message) from None


def _find_comment(records, comment_id, depth=0):
    if depth > 20 or not isinstance(records, list):
        return None
    for record in records:
        if not isinstance(record, dict):
            continue
        if record.get("id") == comment_id:
            return record
        found = _find_comment(record.get("replies", []), comment_id, depth + 1)
        if found is not None:
            return found
    return None


def delete_comment(item_id, comment_id, account, can_moderate=False):
    def updater(data):
        target = _find_comment(_item_comments(data, item_id), comment_id)
        if target is None:
            raise HTTPException(status_code=404, detail="找不到该评论")
        # 在同一锁内检查最新目标并执行软删除，不能使用先前读取的快照。
        if target.get("author") != account and not can_moderate:
            raise HTTPException(status_code=403, detail="无权删除此评论")
        target["isDeleted"] = True
        target["content"] = ""
    db.atomic_update("comments.json", updater, default_data={})


def increment_comment_count(file_name, item_id):
    """派生计数独立更新；失败不能把已提交评论误报为失败或重放评论。"""
    def updater(records):
        for record in records:
            if record.get("id") == item_id:
                old_count = record.get("comments", 0)
                record["comments"] = (old_count if type(old_count) is int and old_count >= 0 else 0) + 1
                break
    try:
        db.atomic_update(file_name, updater, default_data=[])
        return True
    except Exception:
        logging.getLogger(__name__).warning("评论已提交，派生计数暂未更新")
        return False


def _visible_count(records, depth=0):
    if not isinstance(records, list) or depth > 20:
        raise ValueError("Invalid comment thread")
    count = 0
    for record in records:
        if not isinstance(record, dict) or not isinstance(record.get("id"), str) or not record["id"].strip():
            raise ValueError("Invalid comment record")
        deleted = record.get("isDeleted", False)
        if type(deleted) is not bool:
            raise ValueError("Invalid comment state")
        if not deleted:
            count += 1
        count += _visible_count(record.get("replies", []), depth + 1)
    return count


def _validate_comment_text(records):
    for record in records:
        if not isinstance(record.get("author"), str) or not isinstance(record.get("content"), str):
            raise ValueError("Invalid comment text")
        _validate_comment_text(record.get("replies", []))


def read_comment_thread(item_id):
    """严格读取评论快照；存储故障不冒充成功空列表，不改历史主记录。"""
    try:
        snapshot = db.load_data("comments.json", default_data={}, skip_cache=True, strict=True)
        records = _item_comments(snapshot, item_id)
        _visible_count(records)
        _validate_comment_text(records)
        return copy.deepcopy(records)
    except Exception:
        logging.getLogger(__name__).warning("评论列表无法核实")
        raise HTTPException(status_code=503, detail="评论暂时无法读取") from None


def _read_comment_view(item_id):
    records = read_comment_thread(item_id)
    users = db.load_data("users.json", default_data={})
    result = []
    for record in records:
        author_info = users.get(record["author"], {})
        result.append({**record, "author_name": author_info.get("name", record["author"]),
                       "author_avatar": author_info.get("avatarDataUrl", "")})
    return result


async def read_comment_view(item_id):
    """整次评论读取在所属后台工作中执行；繁忙不提交线程池队列。"""
    global _comment_read_pending
    with _comment_read_gate:
        if not _comment_read_accepting or not _comment_read_capacity.acquire(blocking=False):
            raise HTTPException(status_code=503, detail="评论暂时无法读取")
        _comment_read_pending += 1
        _comment_read_drained.clear()
    try:
        return await run_owned_thread(_read_comment_view, item_id)
    except HTTPException:
        raise
    except Exception:
        logging.getLogger(__name__).warning("评论展示无法核实")
        raise HTTPException(status_code=503, detail="评论暂时无法读取") from None
    finally:
        with _comment_read_gate:
            _comment_read_capacity.release()
            _comment_read_pending -= 1
            if not _comment_read_pending:
                _comment_read_drained.set()


def attach_comment_counts(records):
    """只校对响应副本，以评论主记录恢复展示计数，不写内容或资金数据。"""
    try:
        snapshot = db.load_data("comments.json", default_data={}, skip_cache=True, strict=True)
        if not isinstance(snapshot, dict):
            raise ValueError("Invalid comment data")
    except Exception:
        logging.getLogger(__name__).warning("评论计数无法校对，保留已有计数")
        return records
    for record in records:
        try:
            count = _visible_count(snapshot.get(record["id"], []))
        except (KeyError, TypeError, ValueError):
            continue
        record["comments"] = count
    return records