+
+# محرك العلامة الشخصية
+
+**نظام أتمتة العلامة الشخصية بالذكاء الاصطناعي مع 7 وكلاء مستقلين يعملون 24/7**
+
+مبني لـ **سامي محمد العسيري** - مهندس خدمات ميدانية في METCO (أمن المطارات - Smiths Detection)، مطار الملك خالد الدولي، الرياض.
+
+---
+
+## الوكلاء السبعة
+
+| الوكيل | المهمة | الجدول |
+|--------|--------|--------|
+| **وكيل لنكدإن** | نشر محتوى، تفاعل مع الشبكة، تحسين البروفايل | 3 منشورات/أسبوع، 3 تفاعلات/يوم |
+| **وكيل الإيميل** | مراقبة البريد، تصنيف، صياغة ردود | كل 15 دقيقة |
+| **وكيل التواصل الاجتماعي** | نشر تويتر، إعادة صياغة المحتوى | يومياً |
+| **وكيل واتساب** | مساعد شخصي، ردود تلقائية، حجز مواعيد | يعمل دائماً |
+| **محسن السيرة الذاتية** | تحديث السيفي، توليد PDF | شهرياً |
+| **استراتيجي المحتوى** | تحليل الترندات، تقويم محتوى أسبوعي | أسبوعياً + يومياً |
+| **بوت مراقبة الفرص** | يبحث عن وظائف، أخبار، أحداث مهنية | كل ساعتين + ملخص يومي |
+
+## التشغيل السريع
+
+
+
+```bash
+# 1. الإعداد
+cd personal-brand-engine
+make setup
+
+# 2. تعديل المفاتيح
+nano .env
+
+# 3. التشغيل
+make up
+
+# 4. تحميل نموذج الذكاء الاصطناعي
+make pull-model
+
+# 5. التحقق
+make status
+```
+
+
+
+## التكلفة
+
+| الخدمة | التكلفة |
+|--------|---------|
+| GitHub Pages (الصفحة الشخصية) | مجاني |
+| Cal.com (حجز المواعيد) | مجاني |
+| Groq API (الذكاء الاصطناعي) | مجاني |
+| Ollama (ذكاء اصطناعي محلي) | مجاني |
+| واتساب Meta Cloud API | مجاني (1000 محادثة/شهر) |
+| **الإجمالي** | **$0-5/شهر** (السيرفر فقط) |
+
+## الأوامر الأساسية
+
+
+
+```bash
+make help # عرض كل الأوامر
+make up # تشغيل الخدمات
+make down # إيقاف الخدمات
+make logs # عرض السجلات
+make status # فحص الحالة
+```
diff --git a/personal-brand-engine/agents/__init__.py b/personal-brand-engine/agents/__init__.py
new file mode 100644
index 00000000..208c6509
--- /dev/null
+++ b/personal-brand-engine/agents/__init__.py
@@ -0,0 +1,5 @@
+"""Agents package -- autonomous agents for the personal brand engine."""
+
+from agents.base_agent import BaseAgent
+
+__all__ = ["BaseAgent"]
diff --git a/personal-brand-engine/agents/base_agent.py b/personal-brand-engine/agents/base_agent.py
new file mode 100644
index 00000000..ab3d6796
--- /dev/null
+++ b/personal-brand-engine/agents/base_agent.py
@@ -0,0 +1,135 @@
+"""Abstract base class shared by all autonomous agents."""
+
+from __future__ import annotations
+
+import time
+from abc import ABC, abstractmethod
+from typing import Any
+
+from sqlalchemy.orm import Session
+
+from config.settings import (
+ get_brand_profile,
+ get_content_strategy,
+ get_settings,
+)
+from storage.models import AgentLog
+from utils.logger import get_logger
+from utils.notifications import send_notification
+
+logger = get_logger(__name__)
+
+
+class BaseAgent(ABC):
+ """Base class that every agent must inherit from.
+
+ Parameters
+ ----------
+ config:
+ Application :class:`Settings` instance (or a plain dict).
+ llm_client:
+ Any LLM client object the subclass needs (Ollama, Groq, OpenAI, ...).
+ db_session:
+ An active SQLAlchemy :class:`Session`.
+ """
+
+ agent_name: str = "base"
+
+ def __init__(
+ self,
+ config: Any,
+ llm_client: Any,
+ db_session: Session,
+ ) -> None:
+ self.config = config
+ self.llm = llm_client
+ self.db = db_session
+
+ # ------------------------------------------------------------------
+ # Abstract interface
+ # ------------------------------------------------------------------
+
+ @abstractmethod
+ async def run(self, task: str, **kwargs: Any) -> dict:
+ """Execute the agent's primary task and return a result dict.
+
+ Every concrete agent must implement this method.
+ """
+ ...
+
+ # ------------------------------------------------------------------
+ # Shared helpers
+ # ------------------------------------------------------------------
+
+ def log_action(
+ self,
+ action: str,
+ details: str | None = None,
+ *,
+ status: str = "success",
+ duration: float | None = None,
+ ) -> AgentLog:
+ """Persist an :class:`AgentLog` row and emit a structured log line."""
+ entry = AgentLog(
+ agent_name=self.agent_name,
+ task=action,
+ status=status,
+ details=details,
+ duration_seconds=duration,
+ )
+ self.db.add(entry)
+ self.db.flush()
+
+ log_fn = logger.info if status == "success" else logger.error
+ log_fn(
+ "agent_action",
+ agent=self.agent_name,
+ action=action,
+ status=status,
+ duration_seconds=duration,
+ )
+ return entry
+
+ async def notify_owner(self, message: str) -> None:
+ """Send a notification to the project owner.
+
+ Tries Telegram first (if credentials are configured), otherwise
+ falls back to logging the message.
+ """
+ settings = get_settings()
+ await send_notification(message, settings)
+
+ @staticmethod
+ def get_brand_profile() -> dict:
+ """Return the parsed ``brand_profile.yaml`` configuration."""
+ return get_brand_profile()
+
+ @staticmethod
+ def get_content_strategy() -> dict:
+ """Return the parsed ``content_strategy.yaml`` configuration."""
+ return get_content_strategy()
+
+ # ------------------------------------------------------------------
+ # Timing context helper
+ # ------------------------------------------------------------------
+
+ class _Timer:
+ """Minimal wall-clock timer used as a context manager."""
+
+ def __enter__(self) -> "BaseAgent._Timer":
+ self.start = time.perf_counter()
+ return self
+
+ def __exit__(self, *exc: object) -> None:
+ self.elapsed = time.perf_counter() - self.start
+
+ def timer(self) -> _Timer:
+ """Return a context-manager that measures elapsed seconds.
+
+ Usage::
+
+ with self.timer() as t:
+ await do_work()
+ self.log_action("work", duration=t.elapsed)
+ """
+ return self._Timer()
diff --git a/personal-brand-engine/agents/content_strategist/__init__.py b/personal-brand-engine/agents/content_strategist/__init__.py
new file mode 100644
index 00000000..8baa3927
--- /dev/null
+++ b/personal-brand-engine/agents/content_strategist/__init__.py
@@ -0,0 +1,5 @@
+"""Content Strategist agent -- plans content calendars and analyzes trends."""
+
+from agents.content_strategist.agent import ContentStrategistAgent
+
+__all__ = ["ContentStrategistAgent"]
diff --git a/personal-brand-engine/agents/content_strategist/agent.py b/personal-brand-engine/agents/content_strategist/agent.py
new file mode 100644
index 00000000..9e3959e7
--- /dev/null
+++ b/personal-brand-engine/agents/content_strategist/agent.py
@@ -0,0 +1,157 @@
+"""Content Strategist agent -- weekly planning, trend analysis, and calendar management."""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+from sqlalchemy.orm import Session
+
+from agents.base_agent import BaseAgent
+from agents.content_strategist.calendar_planner import (
+ generate_weekly_calendar,
+)
+from agents.content_strategist.trend_analyzer import (
+ analyze_trends,
+)
+from storage.models import ContentCalendar
+
+logger = logging.getLogger(__name__)
+
+# Keywords used for trend scanning (aligned with brand pillars)
+_DEFAULT_KEYWORDS = [
+ "airport security",
+ "aviation safety",
+ "Smiths Detection",
+ "GACA",
+ "X-ray screening",
+ "trace detection",
+ "field services engineering",
+ "Saudi aviation",
+ "ICAO security",
+]
+
+
+class ContentStrategistAgent(BaseAgent):
+ """Autonomous agent that plans Sami's content calendar and tracks trends."""
+
+ agent_name: str = "content_strategist"
+
+ def __init__(
+ self,
+ config: Any,
+ llm_client: Any,
+ db_session: Session,
+ ) -> None:
+ super().__init__(config, llm_client, db_session)
+
+ # ------------------------------------------------------------------
+ # Task dispatcher
+ # ------------------------------------------------------------------
+
+ async def run(self, task: str, **kwargs: Any) -> dict:
+ """Dispatch *task* to the matching handler.
+
+ Supported tasks
+ ---------------
+ - ``weekly_plan`` -- generate a 7-day content calendar
+ - ``trend_analysis`` -- scan for trending topics in aviation security
+ """
+ dispatch = {
+ "weekly_plan": self._weekly_plan,
+ "trend_analysis": self._trend_analysis,
+ }
+
+ handler = dispatch.get(task)
+ if handler is None:
+ self.log_action(task, details=f"Unknown task: {task}", status="failed")
+ return {"status": "error", "message": f"Unknown task: {task}"}
+
+ with self.timer() as t:
+ try:
+ result = await handler(**kwargs)
+ self.log_action(task, details=str(result), duration=t.elapsed)
+ return {"status": "success", "result": result}
+ except Exception as exc:
+ logger.exception("Task %s failed", task)
+ self.log_action(
+ task,
+ details=str(exc),
+ status="failed",
+ duration=t.elapsed,
+ )
+ await self.notify_owner(
+ f"[Content Strategist] Task '{task}' failed: {exc}"
+ )
+ return {"status": "error", "message": str(exc)}
+
+ # ------------------------------------------------------------------
+ # weekly_plan
+ # ------------------------------------------------------------------
+
+ async def _weekly_plan(self, **kwargs: Any) -> dict:
+ """Generate a week of content and persist to the ContentCalendar table."""
+ brand_profile = self.get_brand_profile()
+ content_strategy = self.get_content_strategy()
+
+ # Optionally run trend analysis first to inform the plan
+ trends = kwargs.get("trends")
+ if trends is None:
+ trend_result = await analyze_trends(
+ self.llm,
+ keywords=_DEFAULT_KEYWORDS,
+ brand_profile=brand_profile,
+ )
+ trends = trend_result
+
+ calendar_entries = await generate_weekly_calendar(
+ llm_client=self.llm,
+ brand_profile=brand_profile,
+ content_strategy=content_strategy,
+ trends=trends,
+ )
+
+ # Persist each entry to the database
+ saved_ids: list[int] = []
+ for entry in calendar_entries:
+ row = ContentCalendar(
+ date=datetime.fromisoformat(entry["date"]),
+ pillar=entry["pillar"],
+ topic=entry["topic"],
+ platform=entry["platform"],
+ status="planned",
+ )
+ self.db.add(row)
+ self.db.flush()
+ saved_ids.append(row.id)
+
+ self.db.commit()
+ logger.info("Weekly plan saved: %d entries", len(saved_ids))
+
+ return {
+ "entries_created": len(saved_ids),
+ "calendar_ids": saved_ids,
+ "calendar": calendar_entries,
+ }
+
+ # ------------------------------------------------------------------
+ # trend_analysis
+ # ------------------------------------------------------------------
+
+ async def _trend_analysis(self, **kwargs: Any) -> dict:
+ """Analyze current trends relevant to the brand."""
+ brand_profile = self.get_brand_profile()
+ keywords = kwargs.get("keywords", _DEFAULT_KEYWORDS)
+
+ trends = await analyze_trends(
+ self.llm,
+ keywords=keywords,
+ brand_profile=brand_profile,
+ )
+
+ logger.info("Trend analysis complete: %d trends found", len(trends))
+ return {
+ "trends_found": len(trends),
+ "trends": trends,
+ }
diff --git a/personal-brand-engine/agents/content_strategist/calendar_planner.py b/personal-brand-engine/agents/content_strategist/calendar_planner.py
new file mode 100644
index 00000000..cfa08ffa
--- /dev/null
+++ b/personal-brand-engine/agents/content_strategist/calendar_planner.py
@@ -0,0 +1,233 @@
+"""Calendar planner -- generates a 7-day content plan aligned with brand strategy."""
+
+from __future__ import annotations
+
+import json
+import logging
+from datetime import date, datetime, timedelta, timezone
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# Days of the week when content is posted (0=Mon ... 6=Sun)
+# From schedule.yaml: Sun/Tue/Thu -- ISO weekday: Sun=7, Tue=2, Thu=4
+# Python date.isoweekday(): Mon=1, Tue=2, Wed=3, Thu=4, Fri=5, Sat=6, Sun=7
+_POSTING_DAYS_ISO = {7, 2, 4} # Sunday, Tuesday, Thursday
+
+_SYSTEM_PROMPT = """\
+You are a LinkedIn content strategist for a Field Services Engineer specializing
+in Smiths Detection airport security equipment, based in Riyadh, Saudi Arabia.
+
+Create a 7-day content calendar. Posts are scheduled for Sunday, Tuesday, and Thursday.
+The other days are for engagement-only (likes, comments, networking).
+
+For each posting day, provide:
+- "date": ISO date string (YYYY-MM-DD)
+- "pillar": one of the content pillars from the strategy
+- "topic": specific topic / angle for the post
+- "platform": "linkedin" (primary) or "twitter"
+- "suggested_hook": the opening line / hook for the post (1-2 sentences)
+- "hashtags": list of 3-5 relevant hashtags
+- "content_type": "text", "carousel", "poll", "video_script", or "article"
+
+For non-posting days, include an engagement-only entry:
+- "date": ISO date string
+- "pillar": "engagement"
+- "topic": "Network engagement & community interaction"
+- "platform": "linkedin"
+- "suggested_hook": ""
+- "hashtags": []
+- "content_type": "engagement"
+
+Ensure variety across pillars and content types throughout the week.
+Return ONLY a valid JSON array of 7 objects (one per day).
+"""
+
+
+async def generate_weekly_calendar(
+ llm_client: Any,
+ brand_profile: dict,
+ content_strategy: dict,
+ trends: list[dict] | None = None,
+) -> list[dict]:
+ """Generate a 7-day content plan starting from the next Sunday.
+
+ Parameters
+ ----------
+ llm_client:
+ An :class:`LLMClient` instance.
+ brand_profile:
+ Parsed ``brand_profile.yaml``.
+ content_strategy:
+ Parsed ``content_strategy.yaml``.
+ trends:
+ Optional list of trending topics from :func:`analyze_trends`.
+
+ Returns
+ -------
+ list[dict]
+ Seven entries, one per day, each with ``date``, ``pillar``, ``topic``,
+ ``platform``, ``suggested_hook``, and metadata.
+ """
+ # Calculate the start of next week (next Sunday)
+ today = date.today()
+ days_until_sunday = (7 - today.isoweekday()) % 7
+ if days_until_sunday == 0:
+ days_until_sunday = 7 # If today is Sunday, plan for next week
+ week_start = today + timedelta(days=days_until_sunday)
+
+ week_dates = [week_start + timedelta(days=i) for i in range(7)]
+
+ # Build the pillar descriptions for the prompt
+ pillars = content_strategy.get("content_pillars", [])
+ pillar_text = "\n".join(
+ f"- {p['id']}: {p.get('name_en', '')} -- {p.get('description', '')}"
+ for p in pillars
+ )
+
+ # Format trends
+ trends_text = ""
+ if trends:
+ trends_text = "Trending topics to consider:\n" + "\n".join(
+ f"- {t.get('topic', '')} (relevance: {t.get('relevance', 'medium')}, pillar: {t.get('pillar', '')})"
+ for t in trends[:8]
+ )
+
+ personal = brand_profile.get("personal", {})
+ user_prompt = f"""\
+Generate a 7-day content calendar for the week of {week_start.isoformat()} to {week_dates[-1].isoformat()}.
+
+Professional context:
+- Name: {personal.get('name_en', '')}
+- Role: {personal.get('title_en', '')}
+- Company: {brand_profile.get('employment', {}).get('current', {}).get('company', '')}
+- Location: {personal.get('location_en', '')}
+
+Content pillars:
+{pillar_text}
+
+Posting schedule: Sunday, Tuesday, Thursday
+Engagement-only days: Monday, Wednesday, Saturday, Friday
+
+Tone: {content_strategy.get('tone', {}).get('style', 'professional_approachable')}
+Primary language: Arabic (with English for technical/international content)
+
+{trends_text}
+
+Week dates:
+{chr(10).join(f'- {d.isoformat()} ({d.strftime("%A")})' for d in week_dates)}
+
+Return ONLY a valid JSON array of 7 objects.
+"""
+
+ response = await llm_client.generate(
+ prompt=user_prompt,
+ system_prompt=_SYSTEM_PROMPT,
+ temperature=0.6,
+ max_tokens=2500,
+ )
+
+ calendar = _parse_calendar_response(response.text, week_dates)
+ logger.info("Weekly calendar generated: %d entries", len(calendar))
+ return calendar
+
+
+def _parse_calendar_response(
+ text: str,
+ week_dates: list[date],
+) -> list[dict]:
+ """Parse the LLM response into a calendar list, with fallback generation."""
+ cleaned = text.strip()
+
+ # Strip markdown code fences
+ if cleaned.startswith("```"):
+ first_newline = cleaned.index("\n")
+ cleaned = cleaned[first_newline + 1 :]
+ if cleaned.endswith("```"):
+ cleaned = cleaned[: -len("```")].rstrip()
+
+ try:
+ parsed = json.loads(cleaned)
+ if isinstance(parsed, list):
+ entries = parsed
+ elif isinstance(parsed, dict) and "calendar" in parsed:
+ entries = parsed["calendar"]
+ else:
+ entries = [parsed]
+ except json.JSONDecodeError:
+ logger.warning("Failed to parse calendar JSON; generating fallback")
+ return _generate_fallback_calendar(week_dates)
+
+ # Validate and normalize entries
+ normalized: list[dict] = []
+ for entry in entries:
+ normalized.append(
+ {
+ "date": entry.get("date", ""),
+ "pillar": entry.get("pillar", "engagement"),
+ "topic": entry.get("topic", ""),
+ "platform": entry.get("platform", "linkedin"),
+ "suggested_hook": entry.get("suggested_hook", ""),
+ "hashtags": entry.get("hashtags", []),
+ "content_type": entry.get("content_type", "text"),
+ }
+ )
+
+ # Ensure we have exactly 7 entries (pad with engagement days if needed)
+ while len(normalized) < 7:
+ idx = len(normalized)
+ if idx < len(week_dates):
+ d = week_dates[idx]
+ else:
+ d = week_dates[-1] + timedelta(days=idx - len(week_dates) + 1)
+ normalized.append(
+ {
+ "date": d.isoformat(),
+ "pillar": "engagement",
+ "topic": "Network engagement & community interaction",
+ "platform": "linkedin",
+ "suggested_hook": "",
+ "hashtags": [],
+ "content_type": "engagement",
+ }
+ )
+
+ return normalized[:7]
+
+
+def _generate_fallback_calendar(week_dates: list[date]) -> list[dict]:
+ """Generate a basic fallback calendar when LLM parsing fails."""
+ _fallback_pillars = [
+ "tech_insights",
+ "engagement",
+ "field_life",
+ "engagement",
+ "industry_news",
+ "engagement",
+ "engagement",
+ ]
+ _fallback_topics = [
+ "Weekly airport security technology insight",
+ "Network engagement & community interaction",
+ "A day in the life of a Field Services Engineer",
+ "Network engagement & community interaction",
+ "Industry news commentary and analysis",
+ "Network engagement & community interaction",
+ "Network engagement & community interaction",
+ ]
+
+ entries: list[dict] = []
+ for i, d in enumerate(week_dates[:7]):
+ is_posting_day = d.isoweekday() in _POSTING_DAYS_ISO
+ entries.append(
+ {
+ "date": d.isoformat(),
+ "pillar": _fallback_pillars[i] if is_posting_day else "engagement",
+ "topic": _fallback_topics[i] if is_posting_day else "Network engagement & community interaction",
+ "platform": "linkedin",
+ "suggested_hook": "",
+ "hashtags": [],
+ "content_type": "text" if is_posting_day else "engagement",
+ }
+ )
+ return entries
diff --git a/personal-brand-engine/agents/content_strategist/prompts/strategy_prompts.yaml b/personal-brand-engine/agents/content_strategist/prompts/strategy_prompts.yaml
new file mode 100644
index 00000000..bd7d3180
--- /dev/null
+++ b/personal-brand-engine/agents/content_strategist/prompts/strategy_prompts.yaml
@@ -0,0 +1,122 @@
+# ===================================
+# Content Strategist - System Prompts
+# ===================================
+
+weekly_plan:
+ system: |
+ You are a LinkedIn content strategist for Sami Mohammed Assiri, a Field Services
+ Engineer at METCO specializing in Smiths Detection airport security equipment at
+ King Khalid International Airport, Riyadh.
+
+ Your job is to create a weekly content calendar that:
+ 1. Positions Sami as a thought leader in airport security technology
+ 2. Alternates between content pillars for variety
+ 3. Uses storytelling and real-world field experience (without disclosing sensitive security details)
+ 4. Balances Arabic and English content
+ 5. Follows the posting schedule: Sunday, Tuesday, Thursday
+ 6. Includes engagement strategies for non-posting days
+
+ Content pillars:
+ - tech_insights: Deep dives into Smiths Detection equipment, X-Ray technology, trace detection
+ - field_life: Behind-the-scenes at the airport, daily challenges, engineering stories
+ - professional_growth: Certifications, training, career growth in aviation security
+ - industry_news: ICAO, GACA, TSA regulations and industry developments
+
+ Tone: Professional but approachable. Technical but accessible.
+ Avoid: Sensitive security procedures, classified information, criticizing employers.
+
+ user_template: |
+ Generate a 7-day content calendar for the week starting {start_date}.
+
+ Recent trends to consider:
+ {trends_summary}
+
+ Previous week's performance:
+ {last_week_summary}
+
+ Ensure variety across pillars, content types (text, carousel, poll, article),
+ and topics. Each post should have a compelling hook.
+
+trend_analysis:
+ system: |
+ You are a trend analyst for aviation security and airport technology.
+ Analyze the provided news headlines and identify topics that a Field Services
+ Engineer specializing in Smiths Detection equipment could comment on credibly.
+
+ Focus on:
+ - New screening technologies and regulations
+ - GACA (Saudi aviation authority) announcements
+ - Smiths Detection product launches or updates
+ - Airport security best practices
+ - Saudi Vision 2030 aviation sector developments
+ - International aviation security standards (ICAO, TSA)
+
+ Filter out topics that are:
+ - Too sensitive (specific security vulnerabilities)
+ - Not relevant to an airport security equipment engineer
+ - Outdated (more than 2 weeks old)
+
+ Rate each topic's relevance as high/medium/low and suggest a content angle.
+
+ user_template: |
+ Analyze these headlines for trending topics:
+ {headlines}
+
+ Professional context:
+ - Role: {role}
+ - Specialization: {specialization}
+ - Keywords: {keywords}
+
+ Return the top 10 most relevant trends as a JSON array.
+
+post_generation:
+ system: |
+ You are a LinkedIn ghostwriter for Sami Mohammed Assiri, a Field Services Engineer
+ at METCO specializing in Smiths Detection airport security systems.
+
+ Writing style:
+ - Open with a strong hook (question, bold statement, or personal anecdote)
+ - Use short paragraphs (2-3 lines max)
+ - Include a personal insight or lesson learned
+ - End with a question or call-to-action to drive engagement
+ - Use relevant emojis sparingly (1-2 per post, professional ones only)
+ - Mix Arabic and English naturally (Arabic for storytelling, English for technical terms)
+ - Keep posts between 150-300 words for optimal engagement
+ - Never reveal sensitive airport security procedures
+
+ Format:
+ - Hook line (attention-grabbing opener)
+ - Body (3-4 short paragraphs with the main content)
+ - Takeaway (key lesson or insight)
+ - CTA (call-to-action or engaging question)
+ - Hashtags (3-5 relevant tags)
+
+ user_template: |
+ Write a LinkedIn post about: {topic}
+ Content pillar: {pillar}
+ Language: {language}
+ Content type: {content_type}
+ Suggested hook: {suggested_hook}
+
+ Additional context:
+ {context}
+
+engagement_reply:
+ system: |
+ You are drafting a thoughtful comment on a LinkedIn post on behalf of Sami Mohammed
+ Assiri, a Field Services Engineer specializing in airport security technology.
+
+ Comment guidelines:
+ - Add genuine value (share an insight, ask a thoughtful question, or offer a perspective)
+ - Never be generic ("Great post!" or "Thanks for sharing")
+ - Keep it concise (2-4 sentences)
+ - Relate to your expertise in airport security when naturally relevant
+ - Be supportive and professional
+ - Match the language of the original post (Arabic or English)
+
+ user_template: |
+ Original post by {author}:
+ "{post_content}"
+
+ Write a thoughtful comment from Sami's perspective.
+ Sami's relevant expertise: {relevant_expertise}
diff --git a/personal-brand-engine/agents/content_strategist/trend_analyzer.py b/personal-brand-engine/agents/content_strategist/trend_analyzer.py
new file mode 100644
index 00000000..bf5852db
--- /dev/null
+++ b/personal-brand-engine/agents/content_strategist/trend_analyzer.py
@@ -0,0 +1,224 @@
+"""Trend analyzer -- RSS feeds + LLM to identify relevant trending topics."""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any
+from xml.etree import ElementTree
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# RSS feeds relevant to Sami's brand pillars
+# ---------------------------------------------------------------------------
+
+_RSS_FEEDS: list[dict[str, str]] = [
+ {
+ "name": "Aviation Security International",
+ "url": "https://www.asi-mag.com/feed/",
+ "category": "aviation_security",
+ },
+ {
+ "name": "Airport Technology",
+ "url": "https://www.airport-technology.com/feed/",
+ "category": "airport_tech",
+ },
+ {
+ "name": "Security Today",
+ "url": "https://securitytoday.com/rss-feeds/news.aspx",
+ "category": "security_industry",
+ },
+ {
+ "name": "GACA News (Saudi)",
+ "url": "https://gaca.gov.sa/web/en/rss",
+ "category": "gaca",
+ },
+ {
+ "name": "ICAO Newsroom",
+ "url": "https://www.icao.int/Newsroom/Pages/RSS.aspx",
+ "category": "icao",
+ },
+]
+
+# System prompt for the trend-analysis LLM call
+_SYSTEM_PROMPT = """\
+You are a content strategist for a Field Services Engineer specializing in airport
+security equipment (Smiths Detection). Analyze the provided news headlines and identify
+trending topics that are relevant for LinkedIn content creation.
+
+For each trend, return a JSON array of objects with:
+- "topic": concise topic title
+- "relevance": "high" | "medium" | "low"
+- "pillar": one of "tech_insights", "field_life", "professional_growth", "industry_news"
+- "angle": a brief suggestion for how to turn this into engaging LinkedIn content
+- "source": the feed or keyword that surfaced it
+
+Return ONLY a valid JSON array. Limit to the top 10 most relevant trends.
+"""
+
+
+async def analyze_trends(
+ llm_client: Any,
+ keywords: list[str],
+ brand_profile: dict,
+) -> list[dict]:
+ """Scan RSS feeds and use the LLM to identify relevant trending topics.
+
+ Parameters
+ ----------
+ llm_client:
+ An :class:`LLMClient` instance.
+ keywords:
+ Search terms aligned with the brand pillars.
+ brand_profile:
+ Parsed ``brand_profile.yaml`` dict.
+
+ Returns
+ -------
+ list[dict]
+ Each dict contains ``topic``, ``relevance``, ``pillar``, ``angle``, ``source``.
+ """
+ # Step 1: Fetch RSS headlines
+ headlines = await _fetch_rss_headlines()
+
+ # Step 2: Build LLM prompt
+ personal = brand_profile.get("personal", {})
+ user_prompt = f"""\
+Professional context:
+- Name: {personal.get('name_en', '')}
+- Role: {personal.get('title_en', '')}
+- Specialization: Smiths Detection airport security equipment (HI-SCAN, IONSCAN 600, CTX)
+- Keywords of interest: {', '.join(keywords)}
+
+Recent industry headlines:
+{_format_headlines(headlines)}
+
+Identify the top trending topics relevant to this professional's LinkedIn brand.
+Return ONLY a valid JSON array.
+"""
+
+ response = await llm_client.generate(
+ prompt=user_prompt,
+ system_prompt=_SYSTEM_PROMPT,
+ temperature=0.5,
+ max_tokens=2000,
+ )
+
+ trends = _parse_trends_response(response.text)
+ logger.info("Identified %d trends from %d headlines", len(trends), len(headlines))
+ return trends
+
+
+# ---------------------------------------------------------------------------
+# RSS fetching
+# ---------------------------------------------------------------------------
+
+
+async def _fetch_rss_headlines(timeout: float = 15.0) -> list[dict]:
+ """Fetch headlines from all configured RSS feeds.
+
+ Returns a list of dicts with ``title``, ``link``, ``source``, ``published``.
+ Feeds that fail to load are silently skipped.
+ """
+ headlines: list[dict] = []
+
+ async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client:
+ for feed in _RSS_FEEDS:
+ try:
+ resp = await client.get(feed["url"])
+ resp.raise_for_status()
+ items = _parse_rss_xml(resp.text, source=feed["name"])
+ headlines.extend(items)
+ logger.debug("Fetched %d items from %s", len(items), feed["name"])
+ except Exception as exc:
+ logger.warning("RSS fetch failed for %s: %s", feed["name"], exc)
+
+ return headlines
+
+
+def _parse_rss_xml(xml_text: str, source: str) -> list[dict]:
+ """Parse RSS/Atom XML and extract headline items."""
+ items: list[dict] = []
+ try:
+ root = ElementTree.fromstring(xml_text)
+ except ElementTree.ParseError:
+ logger.warning("Failed to parse XML from %s", source)
+ return items
+
+ # Standard RSS 2.0
+ for item in root.iter("item"):
+ title_el = item.find("title")
+ link_el = item.find("link")
+ pub_el = item.find("pubDate")
+ if title_el is not None and title_el.text:
+ items.append(
+ {
+ "title": title_el.text.strip(),
+ "link": link_el.text.strip() if link_el is not None and link_el.text else "",
+ "source": source,
+ "published": pub_el.text.strip() if pub_el is not None and pub_el.text else "",
+ }
+ )
+
+ # Atom feeds (namespace-aware)
+ atom_ns = "{http://www.w3.org/2005/Atom}"
+ for entry in root.iter(f"{atom_ns}entry"):
+ title_el = entry.find(f"{atom_ns}title")
+ link_el = entry.find(f"{atom_ns}link")
+ pub_el = entry.find(f"{atom_ns}published") or entry.find(f"{atom_ns}updated")
+ if title_el is not None and title_el.text:
+ link_href = ""
+ if link_el is not None:
+ link_href = link_el.get("href", link_el.text or "")
+ items.append(
+ {
+ "title": title_el.text.strip(),
+ "link": link_href.strip() if link_href else "",
+ "source": source,
+ "published": pub_el.text.strip() if pub_el is not None and pub_el.text else "",
+ }
+ )
+
+ return items[:20] # Cap per feed to keep prompt manageable
+
+
+# ---------------------------------------------------------------------------
+# Formatting & parsing
+# ---------------------------------------------------------------------------
+
+
+def _format_headlines(headlines: list[dict]) -> str:
+ """Format headlines into a numbered list for the LLM prompt."""
+ if not headlines:
+ return "(No headlines fetched -- generate trends based on domain knowledge.)"
+ lines: list[str] = []
+ for i, h in enumerate(headlines[:50], start=1): # Cap at 50 total
+ lines.append(f"{i}. [{h['source']}] {h['title']}")
+ return "\n".join(lines)
+
+
+def _parse_trends_response(text: str) -> list[dict]:
+ """Extract a JSON array of trends from the LLM response."""
+ cleaned = text.strip()
+
+ # Strip markdown code fences
+ if cleaned.startswith("```"):
+ first_newline = cleaned.index("\n")
+ cleaned = cleaned[first_newline + 1 :]
+ if cleaned.endswith("```"):
+ cleaned = cleaned[: -len("```")].rstrip()
+
+ try:
+ parsed = json.loads(cleaned)
+ if isinstance(parsed, list):
+ return parsed
+ # Some models wrap in an object
+ if isinstance(parsed, dict) and "trends" in parsed:
+ return parsed["trends"]
+ return [parsed]
+ except json.JSONDecodeError:
+ logger.warning("Failed to parse trend analysis JSON; returning empty list")
+ return []
diff --git a/personal-brand-engine/agents/cv_optimizer/__init__.py b/personal-brand-engine/agents/cv_optimizer/__init__.py
new file mode 100644
index 00000000..69ece90d
--- /dev/null
+++ b/personal-brand-engine/agents/cv_optimizer/__init__.py
@@ -0,0 +1,5 @@
+"""CV Optimizer agent -- enhances and generates professional CVs."""
+
+from agents.cv_optimizer.agent import CVOptimizerAgent
+
+__all__ = ["CVOptimizerAgent"]
diff --git a/personal-brand-engine/agents/cv_optimizer/agent.py b/personal-brand-engine/agents/cv_optimizer/agent.py
new file mode 100644
index 00000000..d46d3303
--- /dev/null
+++ b/personal-brand-engine/agents/cv_optimizer/agent.py
@@ -0,0 +1,132 @@
+"""CV Optimizer agent -- reads brand profile, enhances content via LLM, and generates PDFs."""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from sqlalchemy.orm import Session
+
+from agents.base_agent import BaseAgent
+from agents.cv_optimizer.formatter import (
+ generate_pdf,
+ render_cv_html,
+)
+from agents.cv_optimizer.updater import enhance_cv_content
+
+logger = logging.getLogger(__name__)
+
+# Resolve once so every helper can rely on the path.
+_OUTPUT_DIR = Path(__file__).resolve().parents[2] / "generated_cvs"
+_TEMPLATE_DIR = Path(__file__).resolve().parent / "templates"
+
+
+class CVOptimizerAgent(BaseAgent):
+ """Autonomous agent that keeps Sami's CV polished and up-to-date."""
+
+ agent_name: str = "cv_optimizer"
+
+ def __init__(
+ self,
+ config: Any,
+ llm_client: Any,
+ db_session: Session,
+ ) -> None:
+ super().__init__(config, llm_client, db_session)
+ _OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
+
+ # ------------------------------------------------------------------
+ # Task dispatcher
+ # ------------------------------------------------------------------
+
+ async def run(self, task: str, **kwargs: Any) -> dict:
+ """Dispatch *task* to the matching handler.
+
+ Supported tasks
+ ---------------
+ - ``update_cv`` -- enhance CV content using the LLM
+ - ``generate_pdf`` -- render and export the CV as a PDF
+ """
+ dispatch = {
+ "update_cv": self._update_cv,
+ "generate_pdf": self._generate_pdf,
+ }
+
+ handler = dispatch.get(task)
+ if handler is None:
+ self.log_action(task, details=f"Unknown task: {task}", status="failed")
+ return {"status": "error", "message": f"Unknown task: {task}"}
+
+ with self.timer() as t:
+ try:
+ result = await handler(**kwargs)
+ self.log_action(task, details=str(result), duration=t.elapsed)
+ return {"status": "success", "result": result}
+ except Exception as exc:
+ logger.exception("Task %s failed", task)
+ self.log_action(
+ task,
+ details=str(exc),
+ status="failed",
+ duration=t.elapsed,
+ )
+ await self.notify_owner(
+ f"[CV Optimizer] Task '{task}' failed: {exc}"
+ )
+ return {"status": "error", "message": str(exc)}
+
+ # ------------------------------------------------------------------
+ # update_cv
+ # ------------------------------------------------------------------
+
+ async def _update_cv(self, **kwargs: Any) -> dict:
+ """Use the LLM to enhance CV descriptions and keywords."""
+ brand_profile = self.get_brand_profile()
+
+ enhanced = await enhance_cv_content(self.llm, brand_profile)
+
+ logger.info("CV content enhanced successfully")
+ return {
+ "enhanced": True,
+ "sections_updated": list(enhanced.keys()),
+ }
+
+ # ------------------------------------------------------------------
+ # generate_pdf
+ # ------------------------------------------------------------------
+
+ async def _generate_pdf(self, *, language: str = "en", **kwargs: Any) -> dict:
+ """Render the CV to HTML then convert to PDF.
+
+ Parameters
+ ----------
+ language:
+ ``"en"`` (default) or ``"ar"`` to select the template.
+ """
+ brand_profile = self.get_brand_profile()
+
+ # Optionally enhance first
+ enhanced_profile = await enhance_cv_content(self.llm, brand_profile)
+
+ template_name = f"cv_template_{language}.html"
+ template_path = _TEMPLATE_DIR / template_name
+
+ if not template_path.exists():
+ raise FileNotFoundError(f"Template not found: {template_path}")
+
+ html = render_cv_html(enhanced_profile, str(template_path), language=language)
+
+ timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
+ filename = f"sami_assiri_cv_{language}_{timestamp}.pdf"
+ output_path = _OUTPUT_DIR / filename
+
+ pdf_path = generate_pdf(html, str(output_path))
+
+ logger.info("CV PDF generated: %s", pdf_path)
+ return {
+ "pdf_path": str(pdf_path),
+ "language": language,
+ "filename": filename,
+ }
diff --git a/personal-brand-engine/agents/cv_optimizer/formatter.py b/personal-brand-engine/agents/cv_optimizer/formatter.py
new file mode 100644
index 00000000..a08d1667
--- /dev/null
+++ b/personal-brand-engine/agents/cv_optimizer/formatter.py
@@ -0,0 +1,186 @@
+"""CV rendering and PDF generation -- Jinja2 templates + WeasyPrint."""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+from typing import Any
+
+from jinja2 import BaseLoader, Environment, FileSystemLoader
+
+logger = logging.getLogger(__name__)
+
+
+def render_cv_html(
+ brand_profile: dict,
+ template_path: str,
+ language: str = "en",
+) -> str:
+ """Render a CV as HTML from the Jinja2 template.
+
+ Parameters
+ ----------
+ brand_profile:
+ Full (optionally enhanced) brand profile dict.
+ template_path:
+ Absolute path to the ``.html`` Jinja2 template file.
+ language:
+ ``"en"`` or ``"ar"`` -- passed into the template context.
+
+ Returns
+ -------
+ str
+ Fully rendered HTML string.
+ """
+ tpl_path = Path(template_path)
+ tpl_dir = str(tpl_path.parent)
+ tpl_name = tpl_path.name
+
+ env = Environment(
+ loader=FileSystemLoader(tpl_dir),
+ autoescape=True,
+ )
+ template = env.get_template(tpl_name)
+
+ # Build the template context from the profile
+ context = _build_template_context(brand_profile, language)
+
+ html = template.render(**context)
+ logger.info("CV HTML rendered (%s chars, lang=%s)", len(html), language)
+ return html
+
+
+def generate_pdf(html_content: str, output_path: str) -> Path:
+ """Convert rendered HTML to a PDF file using WeasyPrint.
+
+ Parameters
+ ----------
+ html_content:
+ The full HTML string to convert.
+ output_path:
+ Destination file path for the generated PDF.
+
+ Returns
+ -------
+ Path
+ The path to the written PDF file.
+ """
+ from weasyprint import HTML
+
+ out = Path(output_path)
+ out.parent.mkdir(parents=True, exist_ok=True)
+
+ HTML(string=html_content).write_pdf(str(out))
+ logger.info("PDF generated: %s (%.1f KB)", out, out.stat().st_size / 1024)
+ return out
+
+
+# ---------------------------------------------------------------------------
+# Private helpers
+# ---------------------------------------------------------------------------
+
+
+def _build_template_context(profile: dict, language: str) -> dict:
+ """Flatten the nested profile dict into a template-friendly context."""
+ personal = profile.get("personal", {})
+ employment = profile.get("employment", {})
+ education = profile.get("education", {})
+ enhanced = profile.get("enhanced", {})
+ lang_suffix = f"_{language}"
+
+ # Use enhanced summary if available, otherwise fall back to bio
+ summary = enhanced.get(f"summary_{language}", "") or personal.get(
+ f"bio_{language}", ""
+ )
+
+ # Current role bullets: prefer enhanced, fall back to raw description
+ current = employment.get("current", {})
+ current_bullets = enhanced.get("current_role_bullets", [])
+ if not current_bullets:
+ raw_desc = current.get(f"description_{language}", "")
+ current_bullets = [
+ line.strip().lstrip("- ")
+ for line in raw_desc.strip().splitlines()
+ if line.strip()
+ ]
+
+ # Previous roles: prefer enhanced
+ previous_roles_enhanced = enhanced.get("previous_roles", [])
+ previous_roles_raw = employment.get("previous", [])
+ if previous_roles_enhanced:
+ previous_roles = previous_roles_enhanced
+ else:
+ previous_roles = [
+ {
+ "company": r.get("company", ""),
+ "title": r.get("title", ""),
+ "period": r.get("period", ""),
+ "bullets": r.get("highlights", []),
+ }
+ for r in previous_roles_raw
+ ]
+
+ # Leadership: prefer enhanced
+ leadership_enhanced = enhanced.get("leadership", [])
+ leadership_raw = profile.get("leadership", [])
+ if leadership_enhanced:
+ leadership = leadership_enhanced
+ else:
+ leadership = [
+ {
+ "role": entry.get("role", ""),
+ "organization": entry.get("organization", ""),
+ "period": entry.get("period", ""),
+ "bullets": entry.get("highlights", []),
+ }
+ for entry in leadership_raw
+ ]
+
+ # Skills grouped by category
+ skills = profile.get("skills", {})
+ skill_categories = []
+ _category_labels = {
+ "data_analytics": {"en": "Data Analytics", "ar": "تحليل البيانات"},
+ "project_management": {"en": "Project Management", "ar": "إدارة المشاريع"},
+ "engineering": {"en": "Engineering", "ar": "الهندسة"},
+ "leadership": {"en": "Leadership", "ar": "القيادة"},
+ "languages": {"en": "Languages", "ar": "اللغات"},
+ }
+ for cat_key, items in skills.items():
+ label = _category_labels.get(cat_key, {}).get(language, cat_key.replace("_", " ").title())
+ if cat_key == "languages":
+ formatted_items = [
+ f"{lang['name']} ({lang['level']})" for lang in items
+ ]
+ else:
+ formatted_items = list(items)
+ skill_categories.append({"name": label, "items": formatted_items})
+
+ return {
+ "language": language,
+ "name": personal.get(f"name_{language}", personal.get("name_en", "")),
+ "title": personal.get(f"title_{language}", personal.get("title_en", "")),
+ "headline": personal.get(f"headline_{language}", ""),
+ "email": personal.get("email", ""),
+ "phone": personal.get("phone", ""),
+ "location": personal.get(f"location_{language}", ""),
+ "linkedin": profile.get("links", {}).get("linkedin", ""),
+ "summary": summary,
+ "current_company": current.get("company", current.get("company_ar", "")),
+ "current_title": current.get("title", current.get("title_ar", "")),
+ "current_location": current.get("location", current.get("location_ar", "")),
+ "current_start_date": current.get("start_date", ""),
+ "current_bullets": current_bullets,
+ "previous_roles": previous_roles,
+ "leadership": leadership,
+ "education_degree": education.get("degree", ""),
+ "education_institution": education.get("institution", ""),
+ "education_location": education.get("location", ""),
+ "education_period": education.get("period", ""),
+ "education_highlights": education.get("highlights", []),
+ "certifications": profile.get("certifications", []),
+ "awards": profile.get("awards", []),
+ "skill_categories": skill_categories,
+ "ats_keywords": enhanced.get("skills_keywords", []),
+ "references": profile.get("references", []),
+ }
diff --git a/personal-brand-engine/agents/cv_optimizer/templates/cv_template_ar.html b/personal-brand-engine/agents/cv_optimizer/templates/cv_template_ar.html
new file mode 100644
index 00000000..fc66df10
--- /dev/null
+++ b/personal-brand-engine/agents/cv_optimizer/templates/cv_template_ar.html
@@ -0,0 +1,382 @@
+
+
+
+
+
+
+
+
+
+
+ الملخص المهني
+ {{ summary }}
+
+
+
+
+ الخبرة العملية
+
+
+
+
+
{{ current_company }} — {{ current_location }}
+
+ {% for bullet in current_bullets %}
+ - {{ bullet }}
+ {% endfor %}
+
+
+
+
{{ role.company }}
+
+ {% for bullet in role.bullets %}
+ - {{ bullet }}
+ {% endfor %}
+
+
+
+
+ {% if leadership %}
+
+ القيادة والعمل التطوعي
+ {% for entry in leadership %}
+
+
+
+ {% for bullet in entry.bullets %}
+ - {{ bullet }}
+ {% endfor %}
+
+
+ {% endif %}
+
+
+
+ التعليم
+
+ {{ education_institution }} — {{ education_location }}
+ {% if education_highlights %}
+
+ {% for h in education_highlights %}
+ - {{ h }}
+ {% endfor %}
+
+ {% endif %}
+
+
+
+ {% if certifications %}
+
+ الشهادات المهنية
+
+ {% for cert in certifications %}
+ - {{ cert }}
+ {% endfor %}
+
+
+ {% endif %}
+
+
+ {% if skill_categories %}
+
+ المهارات
+
+ {% for cat in skill_categories %}
+
+
{{ cat.name }}
+
+ {% for item in cat.items %}
+ - {{ item }}
+ {% endfor %}
+
+
+ {% endfor %}
+
+ {% endif %}
+
+
+ {% if awards %}
+
+ الجوائز والتكريمات
+
+ {% for award in awards %}
+ - {{ award }}
+ {% endfor %}
+
+
+ {% endif %}
+
+
+ {% if references %}
+
+ المراجع
+ {% for ref in references %}
+ {{ ref }}
+ {% endfor %}
+
+ {% endif %}
+
+
+ {% if ats_keywords %}
+
{{ ats_keywords | join(', ') }}
+ {% endif %}
+
+
+
+
+
+
+
+
+ Professional Summary
+ {{ summary }}
+
+
+
+
+ Work Experience
+
+
+
+
+
{{ current_company }} — {{ current_location }}
+
+ {% for bullet in current_bullets %}
+ - {{ bullet }}
+ {% endfor %}
+
+
+
+
{{ role.company }}
+
+ {% for bullet in role.bullets %}
+ - {{ bullet }}
+ {% endfor %}
+
+
+
+
+ {% if leadership %}
+
+ Leadership & Volunteering
+ {% for entry in leadership %}
+
+
+
+ {% for bullet in entry.bullets %}
+ - {{ bullet }}
+ {% endfor %}
+
+
+ {% endif %}
+
+
+
+ Education
+
+ {{ education_institution }} — {{ education_location }}
+ {% if education_highlights %}
+
+ {% for h in education_highlights %}
+ - {{ h }}
+ {% endfor %}
+
+ {% endif %}
+
+
+
+ {% if certifications %}
+
+ Certifications
+
+ {% for cert in certifications %}
+ - {{ cert }}
+ {% endfor %}
+
+
+ {% endif %}
+
+
+ {% if skill_categories %}
+
+ Skills
+
+ {% for cat in skill_categories %}
+
+
{{ cat.name }}
+
+ {% for item in cat.items %}
+ - {{ item }}
+ {% endfor %}
+
+
+ {% endfor %}
+
+ {% endif %}
+
+
+ {% if awards %}
+
+ Awards & Recognition
+
+ {% for award in awards %}
+ - {{ award }}
+ {% endfor %}
+
+
+ {% endif %}
+
+
+ {% if references %}
+
+ References
+ {% for ref in references %}
+ {{ ref }}
+ {% endfor %}
+
+ {% endif %}
+
+
+ {% if ats_keywords %}
+
{{ ats_keywords | join(', ') }}
+ {% endif %}
+
+