Spaces:
Running
Running
| """评论记录的锁内修改；路由负责认证、内容存在性及响应契约。""" | |
| 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 | |