maciejpi/nordabiz
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity
c1a8cb6183
nordabiz/services/social_publisher_service.py
Maciej Pienczyn c1a8cb6183
Some checks are pending
NordaBiz Tests / Unit & Integration Tests (push) Waiting to run
Details
NordaBiz Tests / E2E Tests (Playwright) (push) Blocked by required conditions
Details
NordaBiz Tests / Smoke Tests (Production) (push) Blocked by required conditions
Details
NordaBiz Tests / Send Failure Notification (push) Blocked by required conditions
Details
feat: add tone selector for AI content generation in social publisher 
6 tones: Profesjonalny (default), Przyjazny, Oficjalny, Entuzjastyczny,
Informacyjny, Inspirujący. Tone instruction is appended to the AI prompt.
Dropdown appears next to "Generuj AI" button.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-19 09:36:07 +01:00

697 lines
25 KiB
Python
Raw Blame History

"""
Social Media Publisher Service
==============================
Business logic for creating, editing, scheduling and publishing social media posts.
Supports per-company Facebook configuration via OAuth tokens.
"""
import logging
import os
from datetime import datetime
from typing import Optional, Dict, List, Tuple
from database import SessionLocal, SocialPost, SocialMediaConfig, Company, NordaEvent, OAuthToken
logger = logging.getLogger(__name__)
# Post types supported
POST_TYPES = {
'member_spotlight': 'Poznaj Członka NORDA',
'event_invitation': 'Zaproszenie na Wydarzenie',
'event_recap': 'Relacja z Wydarzenia',
'regional_news': 'Aktualności Regionalne',
'chamber_news': 'Aktualności Izby',
}
# AI Prompt templates for NORDA chamber posts (Polish)
AI_PROMPTS = {
'member_spotlight': """Napisz post na Facebook dla Izby Gospodarczej NORDA Biznes, przedstawiający firmę członkowską:
Firma: {company_name}
Branża: {category}
Miasto: {city}
Opis: {description}
Strona WWW: {website}
Post powinien:
- Zaczynać się od ciepłego nagłówka typu "Poznajcie naszego członka!" lub podobnego
- Opisać czym zajmuje się firma (2-3 zdania)
- Podkreślić wartość tej firmy dla społeczności biznesowej
- Zachęcić do odwiedzenia strony firmy
- Zawierać 3-5 hashtagów (#NordaBiznes #IzbaGospodarcza #Wejherowo + branżowe)
- Być napisany ciepło, z dumą i wspierająco
- Mieć 100-200 słów
- Być po polsku
Odpowiedz WYŁĄCZNIE tekstem postu (z hashtagami na końcu).""",
'event_invitation': """Napisz post na Facebook zapraszający na wydarzenie Izby Gospodarczej NORDA Biznes:
Wydarzenie: {event_title}
Data: {event_date}
Miejsce: {event_location}
Opis: {event_description}
Post powinien:
- Być entuzjastyczny i zachęcający do udziału
- Zawierać najważniejsze informacje (co, kiedy, gdzie)
- Wspomnieć korzyści z udziału
- Zawierać CTA (zachęta do rejestracji/kontaktu)
- Zawierać 3-5 hashtagów (#NordaBiznes #Networking #Wejherowo + tematyczne)
- Mieć 100-150 słów
- Być po polsku
Odpowiedz WYŁĄCZNIE tekstem postu.""",
'event_recap': """Napisz post na Facebook będący relacją z wydarzenia Izby Gospodarczej NORDA Biznes:
Wydarzenie: {event_title}
Data: {event_date}
Miejsce: {event_location}
Tematy: {event_topics}
Liczba uczestników: {attendees_count}
Post powinien:
- Dziękować uczestnikom za obecność
- Podsumować najważniejsze tematy/wnioski
- Wspomnieć atmosferę i wartość spotkania
- Zachęcić do udziału w kolejnych wydarzeniach
- Zawierać 3-5 hashtagów
- Mieć 100-200 słów
- Być po polsku, relacyjny i dziękujący
Odpowiedz WYŁĄCZNIE tekstem postu.""",
'regional_news': """Napisz post na Facebook dla Izby Gospodarczej NORDA Biznes komentujący aktualność regionalną:
Temat: {topic}
Źródło: {source}
Fakty: {facts}
Post powinien:
- Informować o temacie w kontekście biznesowym regionu
- Być informacyjny i ekspercki
- Dodać perspektywę Izby (jak to wpływa na lokalny biznes)
- Zawierać 3-5 hashtagów (#NordaBiznes #Pomorze #Wejherowo + tematyczne)
- Mieć 100-200 słów
- Być po polsku
Odpowiedz WYŁĄCZNIE tekstem postu.""",
'chamber_news': """Napisz post na Facebook z aktualnościami Izby Gospodarczej NORDA Biznes:
Temat: {topic}
Szczegóły: {details}
Post powinien:
- Być oficjalny ale przystępny
- Przekazać najważniejsze informacje
- Zachęcić do interakcji (pytania, komentarze)
- Zawierać 3-5 hashtagów (#NordaBiznes #IzbaGospodarcza + tematyczne)
- Mieć 80-150 słów
- Być po polsku
Odpowiedz WYŁĄCZNIE tekstem postu.""",
}
POST_TONES = {
'professional': {
'label': 'Profesjonalny',
'instruction': 'Pisz w tonie profesjonalnym, rzeczowym i biznesowym. Zachowaj powagę, ale bądź przystępny.',
},
'friendly': {
'label': 'Przyjazny',
'instruction': 'Pisz ciepło i przyjaźnie, jakbyś rozmawiał z dobrym znajomym. Używaj bezpośredniego zwrotu do czytelnika.',
},
'formal': {
'label': 'Oficjalny',
'instruction': 'Pisz formalnie i oficjalnie, jak komunikat prasowy. Zachowaj dystans i powagę instytucji.',
},
'enthusiastic': {
'label': 'Entuzjastyczny',
'instruction': 'Pisz z energią i entuzjazmem! Używaj wykrzykników, pozytywnych emocji i dynamicznego języka.',
},
'informative': {
'label': 'Informacyjny',
'instruction': 'Pisz rzeczowo i informacyjnie, jak artykuł prasowy. Skup się na faktach, danych i konkretach.',
},
'inspiring': {
'label': 'Inspirujący',
'instruction': 'Pisz inspirująco i motywująco. Podkreślaj wartości, wizję i potencjał. Zachęcaj do działania.',
},
}
DEFAULT_TONE = 'professional'
class SocialPublisherService:
"""Service for managing social media posts with per-company FB configuration."""
# ---- CRUD Operations ----
def get_posts(self, status: str = None, post_type: str = None,
publishing_company_id: int = None,
limit: int = 50, offset: int = 0) -> List[SocialPost]:
"""Get posts with optional filters."""
db = SessionLocal()
try:
query = db.query(SocialPost).order_by(SocialPost.created_at.desc())
if status:
query = query.filter(SocialPost.status == status)
if post_type:
query = query.filter(SocialPost.post_type == post_type)
if publishing_company_id:
query = query.filter(SocialPost.publishing_company_id == publishing_company_id)
return query.offset(offset).limit(limit).all()
finally:
db.close()
def get_post(self, post_id: int) -> Optional[SocialPost]:
"""Get single post by ID."""
db = SessionLocal()
try:
return db.query(SocialPost).filter(SocialPost.id == post_id).first()
finally:
db.close()
def create_post(self, post_type: str, content: str, user_id: int,
platform: str = 'facebook', hashtags: str = None,
company_id: int = None, event_id: int = None,
publishing_company_id: int = None,
image_path: str = None, ai_model: str = None,
ai_prompt_template: str = None) -> SocialPost:
"""Create a new post draft."""
db = SessionLocal()
try:
post = SocialPost(
post_type=post_type,
platform=platform,
content=content,
hashtags=hashtags,
image_path=image_path,
company_id=company_id,
event_id=event_id,
publishing_company_id=publishing_company_id,
status='draft',
ai_model=ai_model,
ai_prompt_template=ai_prompt_template,
created_by=user_id,
)
db.add(post)
db.commit()
db.refresh(post)
logger.info(f"Created social post #{post.id} type={post_type} pub_company={publishing_company_id}")
return post
except Exception as e:
db.rollback()
logger.error(f"Failed to create social post: {e}")
raise
finally:
db.close()
def update_post(self, post_id: int, **kwargs) -> Optional[SocialPost]:
"""Update post fields. Only draft/approved posts can be edited."""
db = SessionLocal()
try:
post = db.query(SocialPost).filter(SocialPost.id == post_id).first()
if not post:
return None
if post.status in ('published', 'failed'):
logger.warning(f"Cannot edit post #{post_id} with status={post.status}")
return None
allowed_fields = {'content', 'hashtags', 'image_path', 'post_type',
'company_id', 'event_id', 'publishing_company_id', 'scheduled_at'}
for key, value in kwargs.items():
if key in allowed_fields:
setattr(post, key, value)
post.updated_at = datetime.now()
db.commit()
db.refresh(post)
return post
except Exception as e:
db.rollback()
logger.error(f"Failed to update post #{post_id}: {e}")
raise
finally:
db.close()
def delete_post(self, post_id: int) -> bool:
"""Delete a post (only draft/approved)."""
db = SessionLocal()
try:
post = db.query(SocialPost).filter(SocialPost.id == post_id).first()
if not post:
return False
if post.status == 'published':
logger.warning(f"Cannot delete published post #{post_id}")
return False
db.delete(post)
db.commit()
logger.info(f"Deleted social post #{post_id}")
return True
except Exception as e:
db.rollback()
logger.error(f"Failed to delete post #{post_id}: {e}")
return False
finally:
db.close()
# ---- Workflow ----
def approve_post(self, post_id: int, user_id: int) -> Optional[SocialPost]:
"""Approve a draft post."""
db = SessionLocal()
try:
post = db.query(SocialPost).filter(SocialPost.id == post_id).first()
if not post or post.status != 'draft':
return None
post.status = 'approved'
post.approved_by = user_id
post.updated_at = datetime.now()
db.commit()
db.refresh(post)
logger.info(f"Approved post #{post_id} by user #{user_id}")
return post
finally:
db.close()
def schedule_post(self, post_id: int, scheduled_at: datetime) -> Optional[SocialPost]:
"""Schedule an approved post for future publishing."""
db = SessionLocal()
try:
post = db.query(SocialPost).filter(SocialPost.id == post_id).first()
if not post or post.status not in ('draft', 'approved'):
return None
post.status = 'scheduled'
post.scheduled_at = scheduled_at
post.updated_at = datetime.now()
db.commit()
db.refresh(post)
logger.info(f"Scheduled post #{post_id} for {scheduled_at}")
return post
finally:
db.close()
def _get_publish_token(self, db, publishing_company_id: int) -> Tuple[Optional[str], Optional[SocialMediaConfig]]:
"""Get access token and config for publishing.
Tries OAuth token first (from oauth_tokens table), falls back to
manual token in social_media_config.
"""
config = db.query(SocialMediaConfig).filter(
SocialMediaConfig.platform == 'facebook',
SocialMediaConfig.company_id == publishing_company_id,
SocialMediaConfig.is_active == True
).first()
if not config or not config.page_id:
return None, None
# Try OAuth token first
from oauth_service import OAuthService
oauth = OAuthService()
oauth_token = oauth.get_valid_token(db, publishing_company_id, 'meta', 'facebook')
if oauth_token:
return oauth_token, config
# Fallback to manual token in config
if config.access_token:
return config.access_token, config
return None, None
def publish_post(self, post_id: int) -> Tuple[bool, str]:
"""Publish a post to Facebook immediately.
Returns:
(success: bool, message: str)
"""
db = SessionLocal()
try:
post = db.query(SocialPost).filter(SocialPost.id == post_id).first()
if not post:
return False, "Post nie znaleziony"
if post.status not in ('draft', 'approved', 'scheduled'):
return False, f"Nie można opublikować posta ze statusem: {post.status}"
# Determine which company's FB page to publish on
pub_company_id = post.publishing_company_id
if not pub_company_id:
# Fallback: try legacy global config (no company_id)
config = db.query(SocialMediaConfig).filter(
SocialMediaConfig.platform == 'facebook',
SocialMediaConfig.company_id == None,
SocialMediaConfig.is_active == True
).first()
if config and config.access_token and config.page_id:
access_token = config.access_token
else:
return False, "Nie wybrano firmy publikującej. Ustaw 'Publikuj jako' lub skonfiguruj Facebook."
else:
access_token, config = self._get_publish_token(db, pub_company_id)
if not access_token or not config:
return False, "Facebook nie jest skonfigurowany dla wybranej firmy."
# Build message with hashtags
message = post.content
if post.hashtags:
message += f"\n\n{post.hashtags}"
# Determine if draft mode
published = not config.debug_mode
# Publish via Facebook Graph API
from facebook_graph_service import FacebookGraphService
fb = FacebookGraphService(access_token)
image_path = None
if post.image_path:
base_dir = os.path.dirname(os.path.abspath(__file__))
full_path = os.path.join(os.path.dirname(base_dir), post.image_path)
if os.path.exists(full_path):
image_path = full_path
result = fb.create_post(
page_id=config.page_id,
message=message,
image_path=image_path,
published=published,
)
if result and 'id' in result:
post.status = 'published'
post.published_at = datetime.now()
post.meta_post_id = result['id']
post.meta_response = result
post.updated_at = datetime.now()
db.commit()
mode = "DRAFT (debug)" if config.debug_mode else "LIVE"
logger.info(f"Published post #{post_id} -> FB {result['id']} ({mode})")
return True, f"Post opublikowany ({mode}): {result['id']}"
else:
post.status = 'failed'
post.meta_response = result
post.updated_at = datetime.now()
db.commit()
return False, "Publikacja nie powiodła się. Sprawdź konfigurację Facebook."
except Exception as e:
db.rollback()
logger.error(f"Failed to publish post #{post_id}: {e}")
return False, f"Blad publikacji: {str(e)}"
finally:
db.close()
# ---- AI Content Generation ----
def generate_content(self, post_type: str, context: dict, tone: str = None) -> Tuple[str, str]:
"""Generate post content using AI.
Args:
post_type: One of POST_TYPES keys
context: Dict with template variables
tone: One of POST_TONES keys (default: DEFAULT_TONE)
Returns:
(content: str, ai_model: str)
"""
template = AI_PROMPTS.get(post_type)
if not template:
raise ValueError(f"Unknown post type: {post_type}")
# Fill template with context
try:
prompt = template.format(**context)
except KeyError as e:
raise ValueError(f"Missing context field: {e}")
# Add tone instruction
tone_key = tone if tone in POST_TONES else DEFAULT_TONE
tone_info = POST_TONES[tone_key]
prompt += f"\n\nTONACJA: {tone_info['instruction']}"
# Generate with Gemini
from gemini_service import generate_text
result = generate_text(prompt)
if not result:
raise RuntimeError("AI nie wygenerował treści. Spróbuj ponownie.")
return result.strip(), 'gemini-3-flash'
def generate_hashtags(self, content: str, post_type: str = '') -> Tuple[str, str]:
"""Generate hashtags for given post content using AI.
Returns:
(hashtags: str, ai_model: str)
"""
prompt = f"""Na podstawie poniższej treści posta na Facebook Izby Gospodarczej NORDA Biznes,
wygeneruj 5-8 trafnych hashtagów.
Treść posta:
{content}
Zasady:
- Zawsze uwzględnij #NordaBiznes i #IzbaGospodarcza
- Dodaj hashtagi branżowe i lokalne (#Wejherowo, #Pomorze, #Kaszuby)
- Hashtagi powinny zwiększać zasięg i widoczność posta
- Każdy hashtag zaczynaj od #, oddzielaj spacjami
- Odpowiedz WYŁĄCZNIE hashtagami, nic więcej"""
from gemini_service import generate_text
result = generate_text(prompt)
if not result:
raise RuntimeError("AI nie wygenerował hashtagów. Spróbuj ponownie.")
# Clean up - ensure only hashtags
tags = ' '.join(w for w in result.strip().split() if w.startswith('#'))
return tags, 'gemini-3-flash'
def get_company_context(self, company_id: int) -> dict:
"""Get company data for AI prompt context."""
db = SessionLocal()
try:
company = db.query(Company).filter(Company.id == company_id).first()
if not company:
return {}
category_name = ''
if company.category_id and company.category:
category_name = company.category.name
return {
'company_name': company.name or '',
'category': category_name,
'city': company.address_city or 'Wejherowo',
'description': company.description_full or company.description_short or '',
'website': company.website or '',
}
finally:
db.close()
def get_event_context(self, event_id: int) -> dict:
"""Get event data for AI prompt context."""
db = SessionLocal()
try:
event = db.query(NordaEvent).filter(NordaEvent.id == event_id).first()
if not event:
return {}
return {
'event_title': event.title or '',
'event_date': event.event_date.strftime('%d.%m.%Y %H:%M') if event.event_date else '',
'event_location': event.location or '',
'event_description': event.description or '',
}
finally:
db.close()
# ---- Facebook Config ----
def get_fb_config(self, company_id: int = None) -> Optional[SocialMediaConfig]:
"""Get Facebook configuration for a specific company.
If company_id is None, returns legacy global config (backward compat).
"""
db = SessionLocal()
try:
query = db.query(SocialMediaConfig).filter(
SocialMediaConfig.platform == 'facebook'
)
if company_id is not None:
query = query.filter(SocialMediaConfig.company_id == company_id)
else:
query = query.filter(SocialMediaConfig.company_id == None)
return query.first()
finally:
db.close()
def get_all_fb_configs(self) -> List[SocialMediaConfig]:
"""Get all active Facebook configurations across companies."""
db = SessionLocal()
try:
return db.query(SocialMediaConfig).filter(
SocialMediaConfig.platform == 'facebook',
SocialMediaConfig.is_active == True,
SocialMediaConfig.company_id != None,
).all()
finally:
db.close()
def get_configured_companies(self, company_ids: list = None) -> List[Dict]:
"""Get companies that have an active FB configuration.
Args:
company_ids: If provided, filter to only these companies.
None = all companies (admin).
"""
db = SessionLocal()
try:
query = db.query(SocialMediaConfig).filter(
SocialMediaConfig.platform == 'facebook',
SocialMediaConfig.is_active == True,
SocialMediaConfig.company_id != None,
)
if company_ids is not None:
query = query.filter(SocialMediaConfig.company_id.in_(company_ids))
configs = query.all()
return [
{
'company_id': c.company_id,
'company_name': c.company.name if c.company else f'Firma #{c.company_id}',
'page_name': c.page_name,
'page_id': c.page_id,
'debug_mode': c.debug_mode,
}
for c in configs
]
finally:
db.close()
def save_fb_config(self, company_id: int, page_id: str, page_name: str,
debug_mode: bool, user_id: int,
access_token: str = None) -> SocialMediaConfig:
"""Save/update Facebook configuration for a company."""
db = SessionLocal()
try:
config = db.query(SocialMediaConfig).filter(
SocialMediaConfig.platform == 'facebook',
SocialMediaConfig.company_id == company_id,
).first()
if not config:
config = SocialMediaConfig(platform='facebook', company_id=company_id)
db.add(config)
config.page_id = page_id
config.page_name = page_name
if access_token is not None:
config.access_token = access_token
config.debug_mode = debug_mode
config.is_active = True
config.updated_by = user_id
config.updated_at = datetime.now()
db.commit()
db.refresh(config)
logger.info(f"Saved FB config: company={company_id} page={page_name} debug={debug_mode}")
return config
except Exception as e:
db.rollback()
logger.error(f"Failed to save FB config: {e}")
raise
finally:
db.close()
# ---- Engagement Tracking ----
def refresh_engagement(self, post_id: int) -> Optional[Dict]:
"""Refresh engagement metrics from Facebook for a published post."""
db = SessionLocal()
try:
post = db.query(SocialPost).filter(SocialPost.id == post_id).first()
if not post or not post.meta_post_id:
return None
# Get token from publishing company
access_token = None
if post.publishing_company_id:
access_token, _ = self._get_publish_token(db, post.publishing_company_id)
# Fallback to legacy global config
if not access_token:
config = db.query(SocialMediaConfig).filter(
SocialMediaConfig.platform == 'facebook',
SocialMediaConfig.is_active == True
).first()
if config:
access_token = config.access_token
if not access_token:
return None
from facebook_graph_service import FacebookGraphService
fb = FacebookGraphService(access_token)
engagement = fb.get_post_engagement(post.meta_post_id)
if engagement:
post.engagement_likes = engagement.get('likes', 0)
post.engagement_comments = engagement.get('comments', 0)
post.engagement_shares = engagement.get('shares', 0)
post.engagement_reach = engagement.get('reactions_total', 0)
post.engagement_updated_at = datetime.now()
post.updated_at = datetime.now()
db.commit()
return engagement
return None
finally:
db.close()
# ---- Stats ----
def get_stats(self) -> Dict:
"""Get summary statistics for social publisher dashboard."""
db = SessionLocal()
try:
from sqlalchemy import func
total = db.query(func.count(SocialPost.id)).scalar() or 0
by_status = dict(
db.query(SocialPost.status, func.count(SocialPost.id))
.group_by(SocialPost.status).all()
)
by_type = dict(
db.query(SocialPost.post_type, func.count(SocialPost.id))
.group_by(SocialPost.post_type).all()
)
# Avg engagement for published posts
avg_likes = db.query(func.avg(SocialPost.engagement_likes)).filter(
SocialPost.status == 'published'
).scalar() or 0
avg_comments = db.query(func.avg(SocialPost.engagement_comments)).filter(
SocialPost.status == 'published'
).scalar() or 0
return {
'total': total,
'by_status': by_status,
'by_type': by_type,
'avg_engagement': {
'likes': round(float(avg_likes), 1),
'comments': round(float(avg_comments), 1),
},
}
finally:
db.close()
# Singleton instance
social_publisher = SocialPublisherService()
Reference in New Issue View Git Blame Copy Permalink