#!/usr/bin/env python # -*- coding: utf-8 -*- """ نظام الإنجازات المحفز لمراحل المشروع """ import os import sys import json import streamlit as st import pandas as pd import numpy as np import time from datetime import datetime, timedelta import random # إضافة مسار النظام للوصول للملفات المشتركة sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), "../.."))) # استيراد مكونات قاعدة البيانات try: from database.db_connector import get_connection except ImportError: from utils.helpers import get_connection from utils.helpers import format_time, get_user_info, load_icons class AchievementSystem: """نظام الإنجازات المحفز لمراحل المشروع""" def __init__(self, user_id=None): """تهيئة نظام الإنجازات المحفز""" self.user_id = user_id or 1 # استخدام المستخدم الافتراضي إذا لم يتم توفير معرف المستخدم self.conn = get_connection() self.achievements_path = os.path.join(os.path.dirname(__file__), '..', '..', 'data', 'achievements') os.makedirs(self.achievements_path, exist_ok=True) self.user_data_file = os.path.join(self.achievements_path, f'user_{self.user_id}_achievements.json') self.icons = load_icons() # تحميل بيانات المستخدم self.load_user_data() # تعريف قائمة الإنجازات self.define_achievements() def load_user_data(self): """تحميل بيانات إنجازات المستخدم""" try: if os.path.exists(self.user_data_file): with open(self.user_data_file, 'r', encoding='utf-8') as f: self.user_data = json.load(f) else: # بيانات افتراضية عند عدم وجود ملف self.user_data = { 'user_id': self.user_id, 'total_points': 0, 'level': 1, 'unlocked_achievements': [], 'in_progress_achievements': [], 'last_updated': datetime.now().strftime('%Y-%m-%d %H:%M:%S') } self.save_user_data() except Exception as e: st.error(f"خطأ في تحميل بيانات المستخدم: {e}") self.user_data = { 'user_id': self.user_id, 'total_points': 0, 'level': 1, 'unlocked_achievements': [], 'in_progress_achievements': [], 'last_updated': datetime.now().strftime('%Y-%m-%d %H:%M:%S') } def save_user_data(self): """حفظ بيانات إنجازات المستخدم""" try: with open(self.user_data_file, 'w', encoding='utf-8') as f: json.dump(self.user_data, f, ensure_ascii=False, indent=2) except Exception as e: st.error(f"خطأ في حفظ بيانات المستخدم: {e}") def define_achievements(self): """تعريف قائمة الإنجازات المتاحة""" self.achievements = [ { 'id': 'first_project', 'name': 'بداية الرحلة', 'description': 'قم بإنشاء مشروعك الأول', 'icon': '🏆', 'points': 100, 'category': 'مشاريع', 'difficulty': 'سهل' }, { 'id': 'five_projects', 'name': 'محترف المشاريع', 'description': 'قم بإنشاء خمسة مشاريع', 'icon': '🏅', 'points': 500, 'category': 'مشاريع', 'difficulty': 'متوسط' }, { 'id': 'ten_projects', 'name': 'خبير المشاريع', 'description': 'قم بإنشاء عشرة مشاريع', 'icon': '🎖️', 'points': 1000, 'category': 'مشاريع', 'difficulty': 'صعب' }, { 'id': 'first_document_analysis', 'name': 'المحلل الأول', 'description': 'قم بتحليل مستند للمرة الأولى', 'icon': '📊', 'points': 150, 'category': 'تحليل', 'difficulty': 'سهل' }, { 'id': 'five_document_analysis', 'name': 'محلل متمرس', 'description': 'قم بتحليل خمسة مستندات', 'icon': '📈', 'points': 600, 'category': 'تحليل', 'difficulty': 'متوسط' }, { 'id': 'complete_boq', 'name': 'خبير جداول الكميات', 'description': 'أكمل تحليل جدول كميات كامل', 'icon': '📋', 'points': 300, 'category': 'تحليل', 'difficulty': 'متوسط' }, { 'id': 'risk_analysis', 'name': 'محلل المخاطر', 'description': 'أكمل تحليل مخاطر متقدم', 'icon': '⚠️', 'points': 400, 'category': 'مخاطر', 'difficulty': 'متوسط' }, { 'id': 'ten_risk_identified', 'name': 'متنبئ المخاطر', 'description': 'تعرف على عشرة مخاطر في المشاريع', 'icon': '🔍', 'points': 700, 'category': 'مخاطر', 'difficulty': 'صعب' }, { 'id': 'first_terms_analysis', 'name': 'محلل الشروط', 'description': 'قم بتحليل بنود الشروط والأحكام', 'icon': '📝', 'points': 250, 'category': 'تحليل', 'difficulty': 'متوسط' }, { 'id': 'quick_analysis', 'name': 'محلل سريع', 'description': 'أكمل تحليل مستند في أقل من 5 دقائق', 'icon': '⚡', 'points': 500, 'category': 'كفاءة', 'difficulty': 'صعب' }, { 'id': 'voice_narration', 'name': 'مترجم صوتي', 'description': 'استخدم ميزة الترجمة الصوتية لأول مرة', 'icon': '🎙️', 'points': 200, 'category': 'ترجمة', 'difficulty': 'سهل' }, { 'id': 'multilingual_expert', 'name': 'خبير متعدد اللغات', 'description': 'استخدم الترجمة الصوتية بخمس لغات مختلفة', 'icon': '🌍', 'points': 800, 'category': 'ترجمة', 'difficulty': 'صعب' }, { 'id': 'first_map', 'name': 'مستكشف الخرائط', 'description': 'استخدم ميزة الخريطة التفاعلية لأول مرة', 'icon': '🗺️', 'points': 200, 'category': 'خرائط', 'difficulty': 'سهل' }, { 'id': 'ai_fine_tuning', 'name': 'مدرب الذكاء', 'description': 'قم بتدريب نموذج ذكاء اصطناعي مخصص', 'icon': '🧠', 'points': 1000, 'category': 'ذكاء اصطناعي', 'difficulty': 'خبير' }, { 'id': 'pricing_master', 'name': 'سيد التسعير', 'description': 'أكمل حساب تكلفة مشروع بالكامل', 'icon': '💰', 'points': 500, 'category': 'تسعير', 'difficulty': 'متوسط' } ] def calculate_level(self, points): """حساب مستوى المستخدم بناءً على النقاط""" # صيغة بسيطة لحساب المستوى: كل 1000 نقطة = مستوى واحد level = 1 + int(points / 1000) return level def unlock_achievement(self, achievement_id): """إلغاء قفل إنجاز جديد""" # التحقق من وجود الإنجاز في القائمة achievement = next((a for a in self.achievements if a['id'] == achievement_id), None) if not achievement: return False # التحقق من عدم وجود الإنجاز مسبقاً if achievement_id in [a['id'] for a in self.user_data['unlocked_achievements']]: return False # إزالة الإنجاز من قائمة "قيد التقدم" إذا كان موجوداً self.user_data['in_progress_achievements'] = [ a for a in self.user_data['in_progress_achievements'] if a['id'] != achievement_id ] # إضافة الإنجاز إلى القائمة achievement['unlocked_date'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') self.user_data['unlocked_achievements'].append(achievement) # تحديث النقاط والمستوى self.user_data['total_points'] += achievement['points'] self.user_data['level'] = self.calculate_level(self.user_data['total_points']) # حفظ البيانات self.save_user_data() return achievement def update_achievement_progress(self, achievement_id, progress, total): """تحديث تقدم إنجاز معين""" # التحقق من وجود الإنجاز في القائمة achievement = next((a for a in self.achievements if a['id'] == achievement_id), None) if not achievement: return False # التحقق من عدم وجود الإنجاز في قائمة "تم إلغاء قفله" if achievement_id in [a['id'] for a in self.user_data['unlocked_achievements']]: return False # البحث عن الإنجاز في قائمة "قيد التقدم" in_progress_achievement = next( (a for a in self.user_data['in_progress_achievements'] if a['id'] == achievement_id), None ) if in_progress_achievement: # تحديث التقدم in_progress_achievement['progress'] = progress in_progress_achievement['total'] = total in_progress_achievement['percentage'] = min(100, int((progress / total) * 100)) in_progress_achievement['last_updated'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') else: # إضافة الإنجاز إلى قائمة "قيد التقدم" progress_data = achievement.copy() progress_data['progress'] = progress progress_data['total'] = total progress_data['percentage'] = min(100, int((progress / total) * 100)) progress_data['start_date'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') progress_data['last_updated'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') self.user_data['in_progress_achievements'].append(progress_data) # إذا اكتمل التقدم، قم بإلغاء قفل الإنجاز if progress >= total: return self.unlock_achievement(achievement_id) # حفظ البيانات self.save_user_data() return True def check_and_award_achievements(self, action_type, data=None): """التحقق من ومنح الإنجازات بناءً على إجراءات المستخدم""" try: if action_type == 'create_project': # حساب عدد المشاريع projects_count = self._get_projects_count() # منح إنجازات المشاريع if projects_count == 1: self.unlock_achievement('first_project') elif projects_count == 5: self.unlock_achievement('five_projects') elif projects_count == 10: self.unlock_achievement('ten_projects') # تحديث تقدم الإنجاز self.update_achievement_progress('five_projects', min(projects_count, 5), 5) self.update_achievement_progress('ten_projects', min(projects_count, 10), 10) elif action_type == 'analyze_document': # حساب عدد تحليلات المستندات analysis_count = self._get_document_analysis_count() # منح إنجازات تحليل المستندات if analysis_count == 1: self.unlock_achievement('first_document_analysis') elif analysis_count == 5: self.unlock_achievement('five_document_analysis') # تحديث تقدم الإنجاز self.update_achievement_progress('five_document_analysis', min(analysis_count, 5), 5) # التحقق من الوقت المستغرق للتحليل if data and 'duration_seconds' in data and data['duration_seconds'] < 300: # أقل من 5 دقائق self.unlock_achievement('quick_analysis') elif action_type == 'analyze_boq': self.unlock_achievement('complete_boq') elif action_type == 'analyze_terms': self.unlock_achievement('first_terms_analysis') elif action_type == 'analyze_risks': self.unlock_achievement('risk_analysis') # حساب عدد المخاطر المحددة if data and 'risks_count' in data: risks_count = data['risks_count'] risk_total = self._get_total_risks_identified() new_total = risk_total + risks_count # تحديث تقدم إنجاز "متنبئ المخاطر" self.update_achievement_progress('ten_risk_identified', min(new_total, 10), 10) if new_total >= 10 and risk_total < 10: self.unlock_achievement('ten_risk_identified') elif action_type == 'use_voice_narration': self.unlock_achievement('voice_narration') # حساب عدد اللغات المستخدمة if data and 'language' in data: languages_used = self._get_languages_used() if data['language'] not in languages_used: languages_used.append(data['language']) self._save_languages_used(languages_used) # تحديث تقدم إنجاز "خبير متعدد اللغات" self.update_achievement_progress('multilingual_expert', len(languages_used), 5) if len(languages_used) >= 5: self.unlock_achievement('multilingual_expert') elif action_type == 'use_map': self.unlock_achievement('first_map') elif action_type == 'train_ai_model': self.unlock_achievement('ai_fine_tuning') elif action_type == 'complete_pricing': self.unlock_achievement('pricing_master') except Exception as e: st.error(f"خطأ في التحقق من الإنجازات: {e}") def _get_projects_count(self): """الحصول على عدد المشاريع""" try: cursor = self.conn.cursor() cursor.execute("SELECT COUNT(*) FROM documents WHERE user_id = %s AND type = 'project'", (self.user_id,)) count = cursor.fetchone()[0] cursor.close() return count except Exception: # في حالة عدم وجود جدول أو خطأ في الاتصال، استخدم بيانات تقديرية projects_dir = os.path.join(os.path.dirname(__file__), '..', '..', 'data', 'projects') if os.path.exists(projects_dir): return len([f for f in os.listdir(projects_dir) if os.path.isdir(os.path.join(projects_dir, f))]) return 0 def _get_document_analysis_count(self): """الحصول على عدد تحليلات المستندات""" try: cursor = self.conn.cursor() cursor.execute("SELECT COUNT(*) FROM document_analysis WHERE document_id IN (SELECT id FROM documents WHERE user_id = %s)", (self.user_id,)) count = cursor.fetchone()[0] cursor.close() return count except Exception: # في حالة عدم وجود جدول أو خطأ في الاتصال، استخدم بيانات تقديرية return len(self.user_data['unlocked_achievements']) def _get_total_risks_identified(self): """الحصول على إجمالي عدد المخاطر المحددة""" risks_file = os.path.join(self.achievements_path, f'user_{self.user_id}_risks.json') if os.path.exists(risks_file): try: with open(risks_file, 'r', encoding='utf-8') as f: risks_data = json.load(f) return risks_data.get('total_risks', 0) except Exception: return 0 return 0 def _save_total_risks_identified(self, total): """حفظ إجمالي عدد المخاطر المحددة""" risks_file = os.path.join(self.achievements_path, f'user_{self.user_id}_risks.json') try: with open(risks_file, 'w', encoding='utf-8') as f: json.dump({'total_risks': total}, f, ensure_ascii=False, indent=2) except Exception: pass def _get_languages_used(self): """الحصول على قائمة اللغات المستخدمة""" languages_file = os.path.join(self.achievements_path, f'user_{self.user_id}_languages.json') if os.path.exists(languages_file): try: with open(languages_file, 'r', encoding='utf-8') as f: languages_data = json.load(f) return languages_data.get('languages', []) except Exception: return [] return [] def _save_languages_used(self, languages): """حفظ قائمة اللغات المستخدمة""" languages_file = os.path.join(self.achievements_path, f'user_{self.user_id}_languages.json') try: with open(languages_file, 'w', encoding='utf-8') as f: json.dump({'languages': languages}, f, ensure_ascii=False, indent=2) except Exception: pass def render_achievements_tab(self): """عرض علامة تبويب الإنجازات""" st.markdown("

إنجازاتك

", unsafe_allow_html=True) # عرض مستوى المستخدم والنقاط col1, col2 = st.columns([1, 3]) with col1: st.markdown(f"
المستوى {self.user_data['level']}
", unsafe_allow_html=True) with col2: # حساب النقاط المطلوبة للمستوى التالي next_level_points = (self.user_data['level']) * 1000 current_level_points = (self.user_data['level'] - 1) * 1000 progress = (self.user_data['total_points'] - current_level_points) / (next_level_points - current_level_points) st.markdown(f"
{self.user_data['total_points']} نقطة
", unsafe_allow_html=True) st.progress(progress, text=f"المستوى التالي: {next_level_points} نقطة") # تقسيم الإنجازات إلى مجموعات st.markdown("

الإنجازات المفتوحة

", unsafe_allow_html=True) if not self.user_data['unlocked_achievements']: st.info("لم تقم بفتح أي إنجازات حتى الآن. أكمل المهام للحصول على الإنجازات!") else: # عرض الإنجازات المفتوحة بتنسيق الشبكة cols = st.columns(3) for i, achievement in enumerate(self.user_data['unlocked_achievements']): with cols[i % 3]: self._render_achievement_card(achievement, is_unlocked=True) # عرض الإنجازات قيد التقدم st.markdown("

الإنجازات قيد التقدم

", unsafe_allow_html=True) if not self.user_data['in_progress_achievements']: st.info("ليس لديك أي إنجازات قيد التقدم حالياً.") else: # عرض الإنجازات قيد التقدم for achievement in self.user_data['in_progress_achievements']: self._render_progress_achievement(achievement) # عرض الإنجازات المتاحة st.markdown("

الإنجازات المتاحة

", unsafe_allow_html=True) # فلترة الإنجازات غير المفتوحة وغير قيد التقدم unlocked_ids = [a['id'] for a in self.user_data['unlocked_achievements']] in_progress_ids = [a['id'] for a in self.user_data['in_progress_achievements']] available_achievements = [a for a in self.achievements if a['id'] not in unlocked_ids and a['id'] not in in_progress_ids] if not available_achievements: st.success("رائع! لقد حققت جميع الإنجازات المتاحة.") else: # تقسيم الإنجازات المتاحة حسب الفئات categories = sorted(set(a['category'] for a in available_achievements)) for category in categories: st.markdown(f"
{category}
", unsafe_allow_html=True) category_achievements = [a for a in available_achievements if a['category'] == category] cols = st.columns(3) for i, achievement in enumerate(category_achievements): with cols[i % 3]: self._render_achievement_card(achievement, is_unlocked=False) def _render_achievement_card(self, achievement, is_unlocked): """عرض بطاقة إنجاز""" if is_unlocked: card_class = "achievement-card unlocked" icon_class = "achievement-icon unlocked" title_class = "achievement-name unlocked" points_display = f"{achievement['points']} نقطة" date_display = f"تم الفتح: {achievement.get('unlocked_date', 'غير معروف')}" else: card_class = "achievement-card locked" icon_class = "achievement-icon locked" title_class = "achievement-name locked" points_display = f"{achievement['points']} نقطة" date_display = f"صعوبة: {achievement['difficulty']}" html = f"""
{achievement['icon']}
{achievement['name']}
{achievement['description']}
{points_display} {date_display}
""" st.markdown(html, unsafe_allow_html=True) def _render_progress_achievement(self, achievement): """عرض إنجاز قيد التقدم""" progress = achievement.get('percentage', 0) html = f"""
{achievement['icon']}
{achievement['name']}
{achievement['description']}
{achievement['points']} نقطة
""" st.markdown(html, unsafe_allow_html=True) st.progress(progress / 100, text=f"{progress}% ({achievement.get('progress', 0)}/{achievement.get('total', 1)})") def render_achievements_summary(self): """عرض ملخص الإنجازات في لوحة التحكم""" # حساب الإحصائيات total_achievements = len(self.achievements) unlocked_count = len(self.user_data['unlocked_achievements']) in_progress_count = len(self.user_data['in_progress_achievements']) st.markdown(f"""
الإنجازات
المستوى {self.user_data['level']}
{int((unlocked_count / total_achievements) * 100)}%
{unlocked_count} / {total_achievements}
{unlocked_count}
مفتوحة
{in_progress_count}
قيد التقدم
{self.user_data['total_points']}
نقطة
""", unsafe_allow_html=True) # عرض آخر 3 إنجازات تم فتحها if self.user_data['unlocked_achievements']: st.markdown("
آخر الإنجازات
", unsafe_allow_html=True) recent_achievements = sorted( self.user_data['unlocked_achievements'], key=lambda x: x.get('unlocked_date', ''), reverse=True )[:3] for achievement in recent_achievements: st.markdown(f"""
{achievement['icon']}
{achievement['name']}
{achievement.get('unlocked_date', '')}
+{achievement['points']}
""", unsafe_allow_html=True) def render(self): """عرض واجهة نظام الإنجازات""" st.markdown("

نظام الإنجازات المحفز لمراحل المشروع

", unsafe_allow_html=True) st.markdown("""
نظام الإنجازات يحفزك على إكمال المهام وتحقيق أهداف المشروع من خلال مكافآت وإنجازات قابلة للفتح. اكتسب النقاط وارتقِ بمستواك وافتح إنجازات جديدة كلما تقدمت في استخدام نظام تحليل المناقصات.
""", unsafe_allow_html=True) # عرض صندوق معلومات عند تشغيل الوحدة لأول مرة if not self.user_data['unlocked_achievements'] and not self.user_data['in_progress_achievements']: st.info(""" 👋 مرحباً بك في نظام الإنجازات! استكشف الإنجازات المتاحة وابدأ في تحقيقها عن طريق إكمال المهام في أنحاء النظام المختلفة. كلما حققت المزيد من الإنجازات، حصلت على نقاط أكثر وارتقيت في المستويات. ابدأ الآن بإنشاء مشروع جديد أو تحليل مستند! """) # إنشاء علامات تبويب لعرض محتوى مختلف tab1, tab2, tab3 = st.tabs(["الإنجازات", "المستويات والمكافآت", "الإحصائيات"]) with tab1: self.render_achievements_tab() with tab2: st.markdown("

المستويات والمكافآت

", unsafe_allow_html=True) # عرض معلومات عن نظام المستويات st.markdown("""

نظام المستويات يعتمد على النقاط التي تكتسبها من إنجاز المهام وفتح الإنجازات:

كلما ارتقيت في المستويات، تفتح مكافآت وميزات جديدة في النظام!

""", unsafe_allow_html=True) # عرض قائمة المكافآت st.markdown("

المكافآت المتاحة

", unsafe_allow_html=True) rewards = [ {"level": 2, "name": "قوالب مخصصة", "description": "الوصول إلى قوالب مخصصة للتقارير والتحليلات"}, {"level": 3, "name": "تنبيهات متقدمة", "description": "إعدادات إشعارات متقدمة للمشاريع والمواعيد النهائية"}, {"level": 5, "name": "تحليل معزز", "description": "خيارات إضافية لتحليل المستندات والعقود"}, {"level": 7, "name": "تخصيص متقدم", "description": "خيارات إضافية لتخصيص واجهة النظام والتقارير"}, {"level": 10, "name": "وضع الخبراء", "description": "وضع متقدم مع ميزات خاصة متاحة فقط للمستخدمين المخضرمين"} ] for reward in rewards: status = "متاح" if self.user_data['level'] >= reward['level'] else "مقفل" status_class = "available" if self.user_data['level'] >= reward['level'] else "locked" st.markdown(f"""
المستوى {reward['level']}
{reward['name']}
{reward['description']}
{status}
""", unsafe_allow_html=True) with tab3: st.markdown("

إحصائيات الإنجازات

", unsafe_allow_html=True) # إعداد بيانات للرسم البياني categories = {} for achievement in self.achievements: category = achievement['category'] if category not in categories: categories[category] = {"total": 0, "unlocked": 0} categories[category]["total"] += 1 # حساب الإنجازات المفتوحة لكل فئة for achievement in self.user_data['unlocked_achievements']: category = achievement['category'] if category in categories: categories[category]["unlocked"] += 1 # تحويل البيانات إلى DataFrame df = pd.DataFrame([ { "الفئة": category, "المفتوحة": data["unlocked"], "الإجمالي": data["total"], "النسبة": round((data["unlocked"] / data["total"]) * 100 if data["total"] > 0 else 0) } for category, data in categories.items() ]) # عرض البيانات في جدول st.dataframe( df, column_config={ "النسبة": st.column_config.ProgressColumn( "نسبة الإنجاز", format="%d%%", min_value=0, max_value=100 ) }, hide_index=True ) # عرض معلومات إضافية col1, col2, col3 = st.columns(3) with col1: total_points_possible = sum(a['points'] for a in self.achievements) st.metric( "إجمالي النقاط المحتملة", f"{total_points_possible}", f"{int((self.user_data['total_points'] / total_points_possible) * 100)}%" ) with col2: days_since_first = 0 if self.user_data['unlocked_achievements']: first_date = min([ datetime.strptime(a.get('unlocked_date', datetime.now().strftime('%Y-%m-%d %H:%M:%S')), '%Y-%m-%d %H:%M:%S') for a in self.user_data['unlocked_achievements'] ]) days_since_first = (datetime.now() - first_date).days st.metric("أيام النشاط", f"{days_since_first}") with col3: if self.user_data['unlocked_achievements']: achievements_per_day = round(len(self.user_data['unlocked_achievements']) / max(1, days_since_first), 2) st.metric("معدل الإنجازات اليومي", f"{achievements_per_day}") else: st.metric("معدل الإنجازات اليومي", "0") # إضافة CSS مخصص للصفحة st.markdown(""" """, unsafe_allow_html=True)