164 lines
7.3 KiB
Python
164 lines
7.3 KiB
Python
import json
|
|
import ast
|
|
from typing import Dict, Any
|
|
import time
|
|
import datetime
|
|
from openai import OpenAI
|
|
from langchain_openai import ChatOpenAI
|
|
|
|
today = f'{datetime.datetime.now().year}{datetime.datetime.now().month}{datetime.datetime.now().day}'
|
|
|
|
SYSTEM_PROMPT = """
|
|
تو یک استخراجگر ساختاریافتهی اطلاعات (Information Extractor) برای زبان فارسی هستی. وظیفه تو تبدیل متن به یک دیکشنری پایتون (Python Dictionary) دقیقاً مطابق با ساختار و تعاریف زیر است:
|
|
|
|
### قوانین بنیادین:
|
|
1. فقط بر اساس متن ورودی کار کن و هیچ دانش خارجی اضافه نکن.
|
|
2. خروجی باید "فقط" یک دیکشنری معتبر پایتون باشد؛ بدون هیچ مقدمه، موخره یا توضیح اضافه.
|
|
3. تمامی مقادیر (Values) باید به زبان فارسی باشند.
|
|
4. اگر موردی در متن یافت نشد، مقدار آن را لیست خالی [] یا None قرار بده.
|
|
|
|
### تعاریف عملیاتی ساختار خروجی:
|
|
- **title**: (رشته) بین 4 تا 7 کلمه. خلاصهای دقیق که فقط با واژگان موجود در متن ساخته شده و جهتگیری یا دوراهی اصلی متن را نشان دهد.
|
|
- **central_concepts**: (لیست اشیاء) مفاهیم محوری 2 کلمهای مستقیماً از متن (مضافومضافالیه یا صفتوموصوف؛ بدون حروف عطف).
|
|
- **type**: اگر مفهوم منجر به رفاه و سعادت است "زندگیساز" و اگر مانع خیر است "زندگیسوز".
|
|
- **paragraph_effect**: اگر لحن متن درباره آن مثبت/ترغیبی است "تقویت" و اگر منفی/نهی است "تضعیف".
|
|
- **entities**: (لیست رشته) اسامی خاص اشخاص که صریحاً در متن ذکر شدهاند.
|
|
- **rules**: (لیست رشته) حداکثر 10 قاعده کلی اجتماعی/سیاسی/مذهبی. جملات کوتاه، مستقل و انتزاعی (بدون وابستگی به واژگان خاص متن).
|
|
|
|
### ساختار نهایی خروجی:
|
|
{
|
|
"title": str,
|
|
"central_concepts": [{"concept": str, "type": str, "paragraph_effect": str}],
|
|
"entities": [str],
|
|
"rules": [str]
|
|
}
|
|
"""
|
|
|
|
USER_PROMPT = '''
|
|
متن زیر را بر اساس دستورالعملهای سیستمی تحلیل کن و خروجی را در قالب دیکشنری پایتون ارائه بده:
|
|
|
|
## متن برای تحلیل:
|
|
'''
|
|
|
|
def get_key():
|
|
key = 'aa-YUMjnFkdfIZk1p6XkiWvnw3vGfzVEFNg5A5e2qxOn0iJZgkO' # khamenei_metadata
|
|
return key
|
|
|
|
def get_client():
|
|
url = "https://api.avalai.ir/v1"
|
|
client = OpenAI(
|
|
api_key=get_key(),
|
|
base_url=url,
|
|
)
|
|
return client
|
|
|
|
def llm_request(text, model="gemini-2.5-flash-lite"):
|
|
# print(f'using model: {model}')
|
|
|
|
try:
|
|
messages = [
|
|
{"role": "system", "content": SYSTEM_PROMPT},
|
|
{"role": "user", "content": f"{USER_PROMPT}\n{text}"}]
|
|
|
|
response = client.chat.completions.create(
|
|
messages=messages,
|
|
model=model,
|
|
)
|
|
answer = response.choices[0].message.content
|
|
# messages.append({"role": "assistant", "content": answer})
|
|
except Exception as error:
|
|
with open('./leader-answer/error-in-getting-metadata.txt', mode='a+', encoding='utf-8') as file:
|
|
error_message = f'\n\ntext: {text.strip()}\nerror:{error} \n-------------------------------\n'
|
|
file.write(error_message)
|
|
return 'با عرض پوزش؛ متاسفانه خطایی رخ داده است.'
|
|
return answer
|
|
|
|
def text_to_dict(text: str) -> Dict[str, Any]:
|
|
text = text.replace('\n','')
|
|
text = text.lstrip('```json')
|
|
text = text.lstrip('json')
|
|
text = text.lstrip('```python')
|
|
text = text.rstrip('```')
|
|
text = text.strip()
|
|
try:
|
|
return json.loads(text)
|
|
except (json.JSONDecodeError, TypeError):
|
|
return ast.literal_eval(text)
|
|
|
|
client = get_client()
|
|
models = [ "gemini-2.5-flash-lite", "gpt-4o-mini","deepseek-reasoner"]
|
|
|
|
date = str((datetime.datetime.now())).replace(' ','-').replace(':','').replace('.','-')
|
|
|
|
if __name__ == "__main__":
|
|
|
|
with open('./leader_data/khamenei_messages_4.json', 'r', encoding='utf-8') as file:
|
|
data = json.load(file)
|
|
|
|
start = (datetime.datetime.now())
|
|
|
|
# len_pars = 0
|
|
# for item in data:
|
|
# len_pars += len(item['paragraphs'])
|
|
|
|
test_enteries = []
|
|
all_paragraphs = 0
|
|
for index , entery in enumerate(data, 1):
|
|
# if index > 20:
|
|
# break
|
|
id = entery['id']
|
|
if not id == 60106:
|
|
continue
|
|
print(f'{index}/{len(data)}')
|
|
paragraphs = entery["paragraphs"]
|
|
new_paragraphs = []
|
|
full_text = ''
|
|
for counter, paragraph in enumerate(paragraphs, 1):
|
|
full_text += f'{paragraph["text"]}\n'
|
|
result_data = llm_request(full_text)#gpt-4o
|
|
|
|
llm_answer_data = text_to_dict(result_data)
|
|
|
|
entery["central_concepts"]= llm_answer_data['central_concepts']
|
|
entery["characters"]= llm_answer_data['entities']
|
|
entery["rules"]= llm_answer_data['rules']
|
|
|
|
# for counter, paragraph in enumerate(paragraphs, 1):
|
|
|
|
# paragraph_id = paragraph["paragraph_id"]
|
|
# heading = paragraph["heading"] or ''
|
|
# text = paragraph["text"]
|
|
# sentences = paragraph["sentences"]
|
|
# normalized_sentences = paragraph["normalized_sentences"]
|
|
# keyword_matches = paragraph["keyword_matches"]
|
|
# result_data = llm_request(text)#gpt-4o
|
|
|
|
# llm_answer_data = text_to_dict(result_data)
|
|
|
|
# new_paragraphs.append({
|
|
# "paragraph_id" : paragraph_id,
|
|
# "heading" : heading,
|
|
# "text" : text,
|
|
# "sentences" : sentences,
|
|
# "normalized_sentences" : normalized_sentences,
|
|
# "keyword_matches" : keyword_matches,
|
|
# "paragraph_title": llm_answer_data['title'],
|
|
# "central_concepts": llm_answer_data['central_concepts'],
|
|
# "characters": llm_answer_data['entities'],
|
|
# "rules": llm_answer_data['rules'],
|
|
# })
|
|
|
|
# time.sleep(1)
|
|
|
|
entery['paragraphs'] = new_paragraphs
|
|
test_enteries.append(entery)
|
|
all_paragraphs += len(new_paragraphs)
|
|
|
|
with open('./leader_data/leader-metadata-bayanat-howze-fulltext.json', mode='w', encoding='utf-8') as file:
|
|
result_message = json.dump(test_enteries, file, ensure_ascii=False, indent=2)
|
|
|
|
print('---------------------------------------------')
|
|
print(f'full duration: {(datetime.datetime.now() - start).total_seconds()}')
|
|
print(f'all_paragraphs: {all_paragraphs}')
|
|
print('---------------------------------------------')
|