Spaces:
Running
Running
File size: 6,167 Bytes
c8e5eaa | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 | """提示词主记录与分类的锁内局部修改，价格只在副本上投影。"""
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, {})
|