Spaces:
Running
Running
| """帖子主记录在最新文件上的局部修改；不处理资金或评论跨文件清理。""" | |
| import copy | |
| from publication_identity import same_publication | |
| class PostStoreError(Exception): | |
| def __init__(self, status_code, message): | |
| self.status_code = status_code | |
| self.message = message | |
| super().__init__(message) | |
| class _Unchanged(Exception): | |
| def __init__(self, result): self.result = result | |
| def _records(source): | |
| if not isinstance(source, list): | |
| raise ValueError('invalid post source') | |
| identities = set() | |
| for record in source: | |
| if (not isinstance(record, dict) or not isinstance(record.get('id'), str) | |
| or not record['id'].strip() or record['id'] in identities | |
| or not isinstance(record.get('author'), str) or not record['author'].strip()): | |
| raise ValueError('invalid or ambiguous post record') | |
| identities.add(record['id']) | |
| return source | |
| def _mutate(store, callback): | |
| try: | |
| return store.atomic_update('posts.json', lambda source: callback(_records(source)), default_data=[]) | |
| except _Unchanged as unchanged: | |
| return unchanged.result | |
| except PostStoreError: | |
| raise | |
| except Exception: | |
| # 替换后的同步故障也属于未知结果，不推断为未保存，不自动重放。 | |
| raise PostStoreError(503, '帖子保存结果暂时无法确认，请稍后核对') from None | |
| def publish_post(store, record, replay=False): | |
| saved = copy.deepcopy(record) | |
| try: | |
| _records([saved]) | |
| except ValueError: | |
| raise PostStoreError(400, '帖子发布参数无效') from None | |
| def append(records): | |
| existing = next((item for item in records if item['id'] == saved['id']), None) | |
| if existing is not None: | |
| fields = ('author', 'title', 'content', 'cover_image', 'images', 'is_original', 'post_type', 'video_url') | |
| if replay and same_publication(existing, saved, fields): | |
| raise _Unchanged(copy.deepcopy(existing)) | |
| raise PostStoreError(409, '帖子发布请求与已有记录冲突，请核对后再提交') | |
| records.insert(0, copy.deepcopy(saved)) | |
| return copy.deepcopy(saved) | |
| return _mutate(store, append) | |
| def _owned(records, post_id, account, action): | |
| index = next((index for index, post in enumerate(records) if post['id'] == post_id), None) | |
| if index is None: | |
| raise PostStoreError(404, '帖子不存在') | |
| if records[index]['author'] != account: | |
| raise PostStoreError(403, f'无权{action}他人帖子') | |
| return index | |
| def edit_post(store, post_id, account, changes): | |
| fields = {'title', 'content', 'cover_image', 'images', 'is_original', 'post_type', 'video_url'} | |
| if not isinstance(changes, dict) or set(changes) - fields: | |
| raise PostStoreError(400, '帖子编辑参数无效') | |
| saved = {key: copy.deepcopy(value) for key, value in changes.items() if value is not None} | |
| def edit(records): | |
| index = _owned(records, post_id, account, '修改') | |
| edited = copy.deepcopy(records[index]) | |
| edited.update(copy.deepcopy(saved)) | |
| if 'images' in saved: | |
| edited['images'] = saved['images'][:9] | |
| if saved.get('post_type') == 'video': | |
| edited['images'] = [] | |
| records[index] = edited | |
| return copy.deepcopy(edited) | |
| return _mutate(store, edit) | |
| def delete_post(store, post_id, account): | |
| def remove(records): | |
| index = _owned(records, post_id, account, '删除') | |
| records.pop(index) | |
| return _mutate(store, remove) | |