Spaces:
Running
Running
| """提示词主记录与分类的锁内局部修改，价格只在副本上投影。""" | |
| import copy | |
| from publication_identity import same_publication | |
| import datetime | |
| from purchase_transaction import effective_resource_price | |
| class PromptStoreError(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 _mutate(store, filename, callback, default): | |
| try: | |
| return store.atomic_update(filename, callback, default_data=default) | |
| except _Unchanged as unchanged: | |
| return unchanged.result | |
| except PromptStoreError: | |
| raise | |
| except Exception: | |
| raise PromptStoreError(503, '提示词保存结果暂时无法确认，请稍后核对') from None | |
| def _records(source): | |
| if not isinstance(source, list): raise ValueError('invalid prompt collection') | |
| ids = 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 ids | |
| or not isinstance(record.get('author'), str) or not record['author'].strip()): | |
| raise ValueError('invalid prompt identity') | |
| ids.add(record['id']) | |
| return source | |
| def read_prompts(store): | |
| try: | |
| return copy.deepcopy(_records(store.load_data('prompts.json', default_data=[], strict=True, skip_cache=True))) | |
| except Exception: | |
| raise PromptStoreError(503, '提示词暂时不可读取') from None | |
| def project_price(record): | |
| result = copy.deepcopy(record) | |
| result['price'] = effective_resource_price(result) | |
| if result.get('pending_price') is not None and result.get('pending_price_effective_at'): | |
| when = datetime.datetime.fromisoformat(result['pending_price_effective_at']) | |
| now = datetime.datetime.now(when.tzinfo) if when.tzinfo else datetime.datetime.now() | |
| if now >= when: | |
| result['pending_price'] = None | |
| result['pending_price_effective_at'] = None | |
| return result | |
| def publish_prompt(store, record, replay=False): | |
| saved = copy.deepcopy(record) | |
| try: | |
| _records([saved]) | |
| effective_resource_price(saved) | |
| except Exception: | |
| raise PromptStoreError(400, '提示词发布参数无效') from None | |
| def append(source): | |
| records = _records(source) | |
| existing = next((item for item in records if item['id'] == saved['id']), None) | |
| if existing is not None: | |
| fields = ('author', 'prompt_type', 'category', 'title', 'content', 'prompt_text', 'cover_image', | |
| 'media_type', 'video_url', 'audio_url', 'images', 'is_original', 'price', 'tags') | |
| if replay and same_publication(existing, saved, fields): | |
| raise _Unchanged(copy.deepcopy(existing)) | |
| raise PromptStoreError(409, '提示词发布请求与已有记录冲突，请核对后再提交') | |
| records.insert(0, copy.deepcopy(saved)) | |
| return copy.deepcopy(saved) | |
| return _mutate(store, 'prompts.json', append, []) | |
| def _owned(source, prompt_id, account, action): | |
| records = _records(source) | |
| index = next((index for index, record in enumerate(records) if record['id'] == prompt_id), None) | |
| if index is None: raise PromptStoreError(404, '提示词不存在') | |
| if records[index]['author'] != account: | |
| raise PromptStoreError(403, f'无权{action}他人提示词') | |
| return index | |
| def edit_prompt(store, prompt_id, account, apply): | |
| def edit(source): | |
| index = _owned(source, prompt_id, account, '修改') | |
| edited = project_price(source[index]) | |
| apply(edited) | |
| source[index] = edited | |
| return copy.deepcopy(edited) | |
| return _mutate(store, 'prompts.json', edit, []) | |
| def delete_prompt(store, prompt_id, account): | |
| def remove(source): source.pop(_owned(source, prompt_id, account, '删除')) | |
| return _mutate(store, 'prompts.json', remove, []) | |
| def _categories(source, defaults): | |
| if not isinstance(source, dict): raise ValueError('invalid prompt categories') | |
| result = copy.deepcopy(source) | |
| for kind, values in defaults.items(): | |
| existing = result.setdefault(kind, copy.deepcopy(values)) | |
| if not isinstance(existing, list) or not all(isinstance(name, str) and name.strip() for name in existing): | |
| raise ValueError('invalid prompt category group') | |
| return result | |
| def load_categories(store, defaults): | |
| try: | |
| source = store.load_data('prompt_categories.json', default_data={}, strict=True, skip_cache=True) | |
| completed = _categories(source, defaults) | |
| except Exception: | |
| raise PromptStoreError(503, '提示词分类暂时不可读取') from None | |
| if completed == source: return completed | |
| def initialize(latest): | |
| completed = _categories(latest, defaults) | |
| if completed == latest: raise _Unchanged(completed) | |
| latest.clear(); latest.update(completed) | |
| return copy.deepcopy(completed) | |
| return _mutate(store, 'prompt_categories.json', initialize, {}) | |
| def modify_category(store, defaults, kind, name, *, delete=False, maximum=30): | |
| if kind not in defaults or not isinstance(name, str) or not name.strip(): | |
| raise PromptStoreError(400, '分类参数无效') | |
| def update(source): | |
| completed = _categories(source, defaults) | |
| values = completed[kind] | |
| if delete: | |
| if name not in values: raise PromptStoreError(404, '分类不存在') | |
| values.remove(name) | |
| else: | |
| if name in values: raise PromptStoreError(400, '该分类已存在') | |
| if len(values) >= maximum: raise PromptStoreError(400, f'分类数量已达上限({maximum})') | |
| values.append(name) | |
| source.clear(); source.update(completed) | |
| return copy.deepcopy(values) | |
| return _mutate(store, 'prompt_categories.json', update, {}) | |