import json import ast import os from typing import Dict, Any import time import datetime from openai import OpenAI from langchain_openai import ChatOpenAI from nahj_get_metadata_oss import oss_test import asyncio import sqlite3 import data_model as dm # conn = sqlite3.connect('./db/nahj.db') # cursor = conn.cursor() today = f'{datetime.datetime.now().year}{datetime.datetime.now().month}{datetime.datetime.now().day}' ''' این سورس ، حکمت ها و نامه ها و خطبه های نهج البلاغه به همراه پاراگراف هایشان از فایل جیسون به همراه پرامپت به llm مورد نظر میفرسته و پاسخ که یک دیکشنری پایتونی ( شامل مفاهیم کلیدی، شخصیت ها، عنوان ، آیدی ، قوانین و ... ) هست رو به صورت جیسون ذخیره میکنه فرق این سورس با ورژن 1 در این هست که برای هرکدام از تایتل ها و مفاهیم کلیدی و ... از پرامپت مخصوص خودش استفاده میکنه و در چهار فایل جداگانه ذخیره میکنه ''' # پرامپت های مختلف که هرکدام یکی از (تایتل یا مفاهیم کلیدی یا شخصیت و یا قواعد) رو از متن نهج البلاغه استخراج میکنه SYSTEM_PROMPT_orginal = """ تو یک استخراج‌گر ساختاریافته اطلاعات برای متون فارسی هستی. وظیفه تو تحلیل هر پاراگراف و تولید خروجی دقیق بر اساس تعاریف زیر است. فقط و فقط بر اساس متن ورودی عمل کن و هیچ دانش، تفسیر یا مفهوم خارجی اضافه نکن. ساختار ورودی هر ورودی شامل موارد زیر است: id : شناسه متن paragraphs : لیستی از پاراگراف‌ها که هرکدام شامل: paragraph_id text قوانین بنیادین سختگیرانه: خروجی باید فقط یک دیکشنری معتبر پایتون باشد. هیچ توضیح، مقدمه یا متن اضافی تولید نکن. تمام مقادیر باید به زبان فارسی باشند. اگر داده‌ای در متن وجود نداشت، مقدار آن را [] یا None قرار بده. تعاریف عملیاتی استخراج: 1. title یک رشته بین 4 تا 7 کلمه فقط با واژگان موجود در متن ساخته شود جهت‌گیری، تنش یا دوراهی اصلی متن را نشان دهد 2. central_concepts شامل مفاهیم اصلی، محوری و بسیار مهم هر پاراگراف باشد تعداد آن محدود و فقط شامل مفاهیم با اهمیت بالا باشد هر مفهوم دقیقاً دو کلمه‌ای باشد اسامی خاص و اشخاص به هیچ وجه به عنوان کلیدواژه انتخاب نشوند کلیدواژه ها مستقیماً از متن استخراج شود فقط در قالب: مضاف و مضاف‌الیه صفت و موصوف بدون استفاده از حروف عطف(هرگز هرگز هرگز کلمات کلیدواژه با حرف «و» به هم عطف نشوند(کاملا سختگیرانه)) paragraph_effect برای هر مفهوم مرکزی، احساس متن نسبت به آن باید به صورت طیفی عددی مشخص شود: یک عدد اعشاری یا صحیح در بازه -1 تا +1 +1 → بیشترین میزان تقویت -1 → بیشترین میزان تضعیف هرچه عدد به +1 نزدیک‌تر باشد، مفهوم بیشتر تقویت شده است هرچه عدد به -1 نزدیک‌تر باشد، مفهوم بیشتر تضعیف شده است مقادیر بین این دو (مثلاً 0.2 ، -0.4 ، 0.75) مجاز و نشان‌دهنده شدت نسبی هستند 3. persons فقط شخصیت‌های کاملاً حقیقی (افراد واقعی) نام باید صریحاً و دقیقاً در متن آمده باشد اگر هیچ شخصیت حقیقی وجود نداشت، مقدار آن [] باشد شخصیت‌های فرضی، نمادین یا کلی وارد نشوند 4. rules فقط قواعد بسیار مهم و محوری متن استخراج شوند تعداد قواعد محدود باشد هر قاعده یک جمله کوتاه، مستقل و انتزاعی باشد انواع قواعد: قاعده توصیفی بیان رابطه بین دو مفهوم (موضوع + محمول) قاعده هنجاری بیانگر الزام، بایستگی یا ضرورت معمولاً شامل واژگانی مانند: باید، لازم است، ضروری است، نیازمند است، واجب است، حیاتی است و مانند آن 5. paragraph_type برای هر پاراگراف یکی از موارد زیر را انتخاب کن (مرتبط‌ترین گزینه): خطبه: طلیعه سخن (بسم الله و الحمدلله، خوش آمدگویی، تبریک، تسلیت و ...) اشاره یا مقدمه: (اشاره و توضیحی در رابطه با محتوای بحث و مناسبت آن) تیتر: اگر این پاراگراف یک تیتر یا زیرتیتر یا سوتیتر باشد شعر: محتوای پاراگراف یک مصرع یا بیت شعری است آیه: اگر پاراگراف یک آیه از قرآن باشد حدیث: اگر پاراگراف یک روایت یا حدیث از معصومین باشد ارجاع: پاراگراف ارجاع به منابع و پاورقی بخشی از متن است بدنه: اگر از انواع بالا نباشد ساختار دقیق خروجی مورد انتظار باید لیستی از دیکشنری ها باشد که بازای هر پاراگراف تولید شده باشد و به صورت زیر باشد: { "paragraph_id": str, "title": str, "central_concepts": [ { "concept": str, "paragraph_effect": float } ], "paragraph_type": str, "persons": [str], "rules": [ { "rule": str, "type": "توصیفی" | "هنجاری" } ] } """ SYSTEM_PROMPT_title = """ تو یک استخراج‌گر ساختاریافته اطلاعات برای متون فارسی هستی. وظیفه تو تحلیل هر پاراگراف و تولید خروجی دقیق بر اساس تعاریف زیر است. فقط و فقط بر اساس متن ورودی عمل کن و هیچ دانش، تفسیر یا مفهوم خارجی اضافه نکن. ساختار ورودی هر ورودی شامل موارد زیر است: id : شناسه متن paragraphs : لیستی از پاراگراف‌ها که هرکدام شامل: paragraph_id text قوانین بنیادین سختگیرانه: خروجی باید فقط یک دیکشنری معتبر پایتون باشد. هیچ توضیح، مقدمه یا متن اضافی تولید نکن. تمام مقادیر باید به زبان فارسی باشند. اگر داده‌ای در متن وجود نداشت، مقدار آن را [] یا None قرار بده. تعاریف عملیاتی استخراج: 1. title یک رشته بین 4 تا 7 کلمه فقط با واژگان موجود در متن ساخته شود جهت‌گیری، تنش یا دوراهی اصلی متن را نشان دهد 2. paragraph_type برای هر پاراگراف یکی از موارد زیر را انتخاب کن (مرتبط‌ترین گزینه): خطبه: طلیعه سخن (بسم الله و الحمدلله، خوش آمدگویی، تبریک، تسلیت و ...) اشاره یا مقدمه: (اشاره و توضیحی در رابطه با محتوای بحث و مناسبت آن) تیتر: اگر این پاراگراف یک تیتر یا زیرتیتر یا سوتیتر باشد شعر: محتوای پاراگراف یک مصرع یا بیت شعری است آیه: اگر پاراگراف یک آیه از قرآن باشد حدیث: اگر پاراگراف یک روایت یا حدیث از معصومین باشد ارجاع: پاراگراف ارجاع به منابع و پاورقی بخشی از متن است بدنه: اگر از انواع بالا نباشد ساختار دقیق خروجی مورد انتظار باید لیستی از دیکشنری ها باشد که بازای هر پاراگراف تولید شده باشد و به صورت زیر باشد: { "paragraph_id": str, "title": str, "paragraph_type": str, } """ SYSTEM_PROMPT_central = """ تو یک استخراج‌گر ساختاریافته اطلاعات برای متون فارسی هستی. وظیفه تو تحلیل هر پاراگراف و تولید خروجی دقیق بر اساس تعاریف زیر است. فقط و فقط بر اساس متن ورودی عمل کن و هیچ دانش، تفسیر یا مفهوم خارجی اضافه نکن. ساختار ورودی هر ورودی شامل موارد زیر است: id : شناسه متن paragraphs : لیستی از پاراگراف‌ها که هرکدام شامل: paragraph_id text قوانین بنیادین سختگیرانه: خروجی باید فقط یک دیکشنری معتبر پایتون باشد. هیچ توضیح، مقدمه یا متن اضافی تولید نکن. تمام مقادیر باید به زبان فارسی باشند. اگر داده‌ای در متن وجود نداشت، مقدار آن را [] یا None قرار بده. تعاریف عملیاتی استخراج: 1. central_concepts شامل مفاهیم اصلی، محوری و بسیار مهم هر پاراگراف باشد تعداد آن محدود و فقط شامل مفاهیم با اهمیت بالا باشد هر مفهوم دقیقاً دو کلمه‌ای باشد اسامی خاص و اشخاص به هیچ وجه به عنوان کلیدواژه انتخاب نشوند کلیدواژه ها مستقیماً از متن استخراج شود فقط در قالب: مضاف و مضاف‌الیه صفت و موصوف بدون استفاده از حروف عطف(هرگز هرگز هرگز کلمات کلیدواژه با حرف «و» به هم عطف نشوند(کاملا سختگیرانه)) paragraph_effect برای هر مفهوم مرکزی، احساس متن نسبت به آن باید به صورت طیفی عددی مشخص شود: یک عدد اعشاری یا صحیح در بازه -1 تا +1 +1 → بیشترین میزان تقویت -1 → بیشترین میزان تضعیف هرچه عدد به +1 نزدیک‌تر باشد، مفهوم بیشتر تقویت شده است هرچه عدد به -1 نزدیک‌تر باشد، مفهوم بیشتر تضعیف شده است مقادیر بین این دو (مثلاً 0.2 ، -0.4 ، 0.75) مجاز و نشان‌دهنده شدت نسبی هستند ساختار دقیق خروجی مورد انتظار باید لیستی از دیکشنری ها باشد که بازای هر پاراگراف تولید شده باشد و دقیقا (خیلی مهم) به صورت زیر باشد: [{ "central_concepts": [ { "concept": str, "paragraph_effect": float } ] },...] """ SYSTEM_PROMPT_person = """ تو یک استخراج‌گر ساختاریافته اطلاعات برای متون فارسی هستی. وظیفه تو تحلیل هر پاراگراف و تولید خروجی دقیق بر اساس تعاریف زیر است. فقط و فقط بر اساس متن ورودی عمل کن و هیچ دانش، تفسیر یا مفهوم خارجی اضافه نکن. ساختار ورودی هر ورودی شامل موارد زیر است: id : شناسه متن paragraphs : لیستی از پاراگراف‌ها که هرکدام شامل: paragraph_id text قوانین بنیادین سختگیرانه: خروجی باید فقط یک دیکشنری معتبر پایتون باشد. هیچ توضیح، مقدمه یا متن اضافی تولید نکن. تمام مقادیر باید به زبان فارسی باشند. اگر داده‌ای در متن وجود نداشت، مقدار آن را [] یا None قرار بده. تعاریف عملیاتی استخراج: 1. persons فقط شخصیت‌های کاملاً حقیقی (افراد واقعی) نام باید صریحاً و دقیقاً در متن آمده باشد اگر هیچ شخصیت حقیقی وجود نداشت، مقدار آن [] باشد شخصیت‌های فرضی، نمادین یا کلی وارد نشوند ساختار دقیق خروجی مورد انتظار باید لیستی از دیکشنری ها باشد که بازای هر پاراگراف تولید شده باشد و دقیقا (بسیار مهم) به صورت زیر باشد: [{ "persons": [str] },...] """ SYSTEM_PROMPT_rules = """ تو یک استخراج‌گر ساختاریافته اطلاعات برای متون فارسی هستی. وظیفه تو تحلیل هر پاراگراف و تولید خروجی دقیق بر اساس تعاریف زیر است. فقط و فقط بر اساس متن ورودی عمل کن و هیچ دانش، تفسیر یا مفهوم خارجی اضافه نکن. ساختار ورودی هر ورودی شامل موارد زیر است: id : شناسه متن paragraphs : لیستی از پاراگراف‌ها که هرکدام شامل: paragraph_id text قوانین بنیادین سختگیرانه: خروجی باید فقط یک دیکشنری معتبر پایتون باشد. هیچ توضیح، مقدمه یا متن اضافی تولید نکن. تمام مقادیر باید به زبان فارسی باشند. اگر داده‌ای در متن وجود نداشت، مقدار آن را [] یا None قرار بده. تعاریف عملیاتی استخراج: 1. rules فقط قواعد بسیار مهم و محوری متن استخراج شوند تعداد قواعد محدود باشد هر قاعده یک جمله کوتاه، مستقل و انتزاعی باشد انواع قواعد: قاعده توصیفی بیان رابطه بین دو مفهوم (موضوع + محمول) قاعده هنجاری بیانگر الزام، بایستگی یا ضرورت معمولاً شامل واژگانی مانند: باید، لازم است، ضروری است، نیازمند است، واجب است، حیاتی است و مانند آن ساختار دقیق خروجی مورد انتظار باید لیستی از دیکشنری ها باشد که بازای هر پاراگراف تولید شده باشد و دقیقا (بسیار مهم) به صورت زیر باشد: [{ "rules": [ { "rule": str, "type": "توصیفی" | "هنجاری" } ] },...] """ SYSTEM_PROMPT = SYSTEM_PROMPT_rules prompts = [SYSTEM_PROMPT_title, SYSTEM_PROMPT_central, SYSTEM_PROMPT_person, SYSTEM_PROMPT_rules] outs = ["title","central","person","rules"] USER_PROMPT = ''' متن زیر را بر اساس دستورالعمل‌های سیستمی تحلیل کن و خروجی را در قالب دیکشنری پایتون ارائه بده: ## ساختار جیسون برای تحلیل: ''' def get_key(): key = 'aa-Fu5oeQv8jx8NCWV39WenJ7Yy1mbcFJ4P20CLQURkql2Eleta' # nahj key return key def get_client(): url = "https://api.avalapis.ir/v1" #"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(llm_error_path, mode='a+', encoding='utf-8') as file: error_message = f'\n\ntext: {str(text)}\nerror:{error} \n-------------------------------\n' file.write(error_message) return 'Ooops ... Error!' 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('.','-') def find_passed_data_ids(output_metadata_jsonl_path): passed_data_ids = [] with open(output_metadata_jsonl_path, 'r', encoding='utf-8') as file: passed_data = file.readlines() for pd in passed_data: passed_data_ids.append(str(json.loads(pd)['id'])) return passed_data_ids if __name__ == "__main__": dm.create_tables() # متد ساخت تمام تیبل های مورد نیاز input_data_path = './nahj_data/all_nahj_CONTEXT.json' # شامل تمامی (خطبه و نامه و حکمت ها) به همراه پاراگراف هایشان llm_error_path = './nahj-answer/error-in-getting-metadata-Final.txt' previous_peroid_errors_path = "./nahj_data/error-ids-Final.txt" current_peroid_errors_path = "./nahj_data/error-ids3-Final.txt" output_metadata_jsonl_path = f'./nahj_data/nahj-metadata-jsonline.json' file_path = './nahj_data/nahj-metadata-jsonline.json' # 1. حذف فایل اگر وجود داشته باشد if os.path.exists(file_path): os.remove(file_path) # 2. ایجاد یک فایل خالی جدید با همان نام with open(file_path, 'w') as f: pass output_metadata_json_path = f'./nahj_data/nahj-metadata-TEST.json' with open(input_data_path, 'r', encoding='utf-8') as file: data = json.load(file) passed_data_ids = [] passed_data_ids = find_passed_data_ids(output_metadata_jsonl_path) failed_ids = [] with open(previous_peroid_errors_path, "r", encoding="utf-8") as f: failed = f.read() failed_ids = failed.splitlines() start = (datetime.datetime.now()) print(f'start: {start}') # len_pars = 0 # for item in data: # len_pars += len(item['paragraphs']) error_ids = [] test_enteries = [] all_paragraphs = 0 NN=-1 # این عدد صرفا برای آیدی استفاده میشود period = 1 end = False while True: print(f"******* PERIOD :: {period} *******") for index ,entery in enumerate(data, 1): NN+=1 err = False if index > 799: end = True break id = entery['id'] # خارج کردن داده هایی که قبلا کرول شده if str(id) in passed_data_ids: continue # برای دور دوم به بعد که برخی از شناسه ها به خطر خورده ، شرط زیر از کامنت خارج شود # if not str(id) in failed_ids: # continue # if not id == 27793 # continue print(f'id: {id} - record: {index}/{len(data)} - period: {period}') for path in outs : # این حلقه برای اجرای هر چهار پرامپته که هرکدام یکی از (تایتل ، مفاهیم کلیدی ، شخصیت ها و قواعد ) را استخراج میکنه if err == True : continue elif path == "title": SYSTEM_PROMPT = SYSTEM_PROMPT_title elif path == "central": SYSTEM_PROMPT = SYSTEM_PROMPT_central elif path == "person": SYSTEM_PROMPT = SYSTEM_PROMPT_person elif path == "rules": SYSTEM_PROMPT = SYSTEM_PROMPT_rules llm_answer_data = '' new_entry = {} new_paragraphs = [] new_entry['id'] = id # new_entry['keywords'] = entery['keywords'] paragraphs = entery["paragraphs"] for p in paragraphs: large_title = p['large_title'] new_paragraphs.append({ 'paragraph_id': p['paragraph_id'], 'text': p['text'] # 'text': f"بخشی از {large_title} : {p['text'] }" }) new_entry['paragraphs'] = new_paragraphs try: # result_data = asyncio.run(oss_test(SYSTEM_PROMPT,USER_PROMPT,new_entry))#gpt-4o result_data = llm_request(new_entry) llm_answer_data = text_to_dict(result_data) if path == "title" : entery['paragraph_metadata'] = [] for num ,sec in enumerate(llm_answer_data): entery['paragraph_metadata'].append({ 'paragraph_id': sec['paragraph_id'], 'title': sec['title'], 'paragraph_type': sec['paragraph_type'] }) if len(llm_answer_data) != len(entery['paragraph_metadata']) : print("error!!!!!!!!!!") if entery['id'] not in error_ids: error_ids.append(entery['id']) if path == "central" : for num ,sec in enumerate(llm_answer_data): entery['paragraph_metadata'][num]['central_concepts'] = sec['central_concepts'] if path == "person" : for num ,sec in enumerate(llm_answer_data): entery['paragraph_metadata'][num]['persons'] = sec['persons'] if path == "rules" : for num ,sec in enumerate(llm_answer_data): entery['paragraph_metadata'][num]['rules'] = sec['rules'] except Exception as e: err = True print(f'error id: {id} - {e} >> llm result: {result_data}') if id not in error_ids: error_ids.append(id) with open(current_peroid_errors_path, "a", encoding="utf-8") as f: f.write(f"{id}\n") continue # entery['paragraph_metadata'] = llm_answer_data # در این قسمت متادیتای پارتها استخراج میشود و در تیبل های اسکیو ال لایت ذخیره میشود context_id = id title = entery['title'] large_title = entery['large_title'] url = entery['url'] typee = entery['type'] i_link = entery['interpretation_link'] N=0 for part in paragraphs : id_ = f"num{NN}{N}" text = part['text'] part_id = part['paragraph_id'] arabic_text = part['arabic_text'] ai_title = entery['paragraph_metadata'][N]['title'] paragraph_type = entery['paragraph_metadata'][N]['paragraph_type'] dm.insert_data_to_speechs([id_, context_id, part_id, title, large_title, text, url, typee, arabic_text, i_link, ai_title, paragraph_type]) # cursor.execute("INSERT INTO speeches (id, context_id, part_id, title, large_title, normalized_sentence, url, types, arabic_text, interpretation_links, ai_title, ai_paragraph_type) \ # VALUES (?, ?, ?, ? ,? ,? ,? ,? ,? ,? ,? ,?)", # ([id_, context_id, part_id, title, large_title, text, url, typee, arabic_text, i_link, ai_title, paragraph_type])) # conn.commit() central_concepts = entery['paragraph_metadata'][N]['central_concepts'] persons = entery['paragraph_metadata'][N]['persons'] rules = entery['paragraph_metadata'][N]['rules'] k=0 for row in central_concepts: k+=1 c_id = context_id+part_id+f"c{k}" concept = row['concept'] paragraph_effect = row['paragraph_effect'] dm.insert_data_to_central_concepts([c_id, concept, paragraph_effect, part_id]) # cursor.execute("""INSERT INTO central_concepts (id, concept, paragraph_effect, part_id) # VALUES(?, ?, ?, ?)""",(c_id, concept, paragraph_effect, part_id)) # conn.commit() k=0 for row in persons: k+=1 c_id = context_id+part_id+f"p{k}" person = row dm.insert_data_to_persons([c_id, person, part_id]) # cursor.execute("""INSERT INTO persons (id, person, part_id) # VALUES(?, ?, ?)""",(c_id, person, part_id)) # conn.commit() k=0 for row in rules: k+=1 c_id = context_id+part_id+f"r{k}" rule = row['rule'] rule_type = row['type'] dm.insert_data_to_rules([c_id, rule, rule_type, part_id]) # cursor.execute("""INSERT INTO rules (id, rule, type, part_id) # VALUES(?, ?, ?, ?)""",([c_id, rule, rule_type, part_id])) # conn.commit() N+=1 test_enteries.append(entery) with open(output_metadata_jsonl_path, 'a', encoding='utf-8') as f: json.dump(entery, f, ensure_ascii=False) f.write('\n') time.sleep(1) passed_data_ids = find_passed_data_ids(output_metadata_jsonl_path) if len(data) == len(passed_data_ids): print('ALL DATA PASSED OK!') break print(f'##### period result: passd {len(passed_data_ids)}/{len(data)} #####') period+= 1 if end == True: break # with open(f'./leader_data/leader-metadata-bayanat-{id}.json', mode='w', encoding='utf-8') as file: with open(output_metadata_json_path, mode='w', encoding='utf-8') as file: result_message = json.dump(test_enteries, file, ensure_ascii=False, indent=2) with open("./nahj_data/error_ids_TEST.json", mode='w', encoding='utf-8') as file: result_message = json.dump(error_ids, file, ensure_ascii=False, indent=2) print('all done!') print('---------------------------------------------') print(f'full duration: {(datetime.datetime.now() - start).total_seconds()}') print(f'all_paragraphs: {all_paragraphs}') print('---------------------------------------------')