maciejpi/nordabiz
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

refactor: Migrate recommendations API routes to api blueprint

Browse Source 
- Created blueprints/api/routes_recommendations.py with 4 routes:
  - /api/recommendations/<company_id> (GET)
  - /api/recommendations/create (POST)
  - /api/recommendations/<rec_id>/edit (POST)
  - /api/recommendations/<rec_id>/delete (POST)
- Removed ~320 lines from app.py (7378 -> 7057)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Maciej Pienczyn 2026-01-31 17:39:48 +01:00
parent 3c5795ee4a
commit eaadef0bc1
3 changed files with 351 additions and 319 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

323
app.py
View File

@ -1111,327 +1111,12 @@ def health_full():
# Routes: /api/analytics/track, /api/analytics/heartbeat, /api/analytics/scroll,
# /api/analytics/error, /api/analytics/performance, /api/analytics/conversion
# ============================================================
# RECOMMENDATIONS API ROUTES
# RECOMMENDATIONS API ROUTES - MOVED TO blueprints/api/routes_recommendations.py
# ============================================================
@app.route('/api/recommendations/<int:company_id>', methods=['GET'])
@login_required
def api_get_recommendations(company_id):
"""API: Get all approved recommendations for a company"""
db = SessionLocal()
try:
# Verify company exists
company = db.query(Company).filter_by(id=company_id).first()
if not company:
return jsonify({
'success': False,
'error': 'Firma nie znaleziona'
}), 404
# Query recommendations with user details
recommendations = db.query(CompanyRecommendation).filter_by(
company_id=company_id,
status='approved'
).join(User, CompanyRecommendation.user_id == User.id).order_by(CompanyRecommendation.created_at.desc()).all()
# Build response with recommender details
result = []
for rec in recommendations:
recommender = db.query(User).filter_by(id=rec.user_id).first()
recommender_company = None
if recommender and recommender.company_id:
recommender_company = db.query(Company).filter_by(id=recommender.company_id).first()
rec_data = {
'id': rec.id,
'recommendation_text': rec.recommendation_text,
'service_category': rec.service_category,
'created_at': rec.created_at.isoformat() if rec.created_at else None,
'updated_at': rec.updated_at.isoformat() if rec.updated_at else None,
'recommender': {
'name': recommender.full_name if recommender else '[Użytkownik usunięty]',
'email': recommender.email if (recommender and rec.show_contact) else None,
'phone': recommender.phone if (recommender and rec.show_contact) else None,
'company_id': recommender_company.id if recommender_company else None,
'company_name': recommender_company.name if recommender_company else None,
'company_slug': recommender_company.slug if recommender_company else None
}
}
result.append(rec_data)
return jsonify({
'success': True,
'company_id': company_id,
'company_name': company.name,
'recommendations': result,
'count': len(result)
})
except Exception as e:
logger.error(f"Error fetching recommendations for company {company_id}: {e}")
return jsonify({
'success': False,
'error': 'Wystąpił błąd podczas pobierania rekomendacji'
}), 500
finally:
db.close()
@app.route('/api/recommendations/create', methods=['POST'])
@login_required
def api_create_recommendation():
"""API: Create a new recommendation"""
db = SessionLocal()
try:
# Get JSON data
data = request.get_json()
if not data:
return jsonify({
'success': False,
'error': 'Brak danych'
}), 400
company_id = data.get('company_id')
recommendation_text = data.get('recommendation_text', '').strip()
service_category = data.get('service_category', '').strip() or None
show_contact = data.get('show_contact', True)
# Validate required fields
if not company_id:
return jsonify({
'success': False,
'error': 'Brak ID firmy'
}), 400
if not recommendation_text:
return jsonify({
'success': False,
'error': 'Treść rekomendacji jest wymagana'
}), 400
# Validate text length (50-2000 characters)
if len(recommendation_text) < 50:
return jsonify({
'success': False,
'error': 'Rekomendacja musi mieć co najmniej 50 znaków'
}), 400
if len(recommendation_text) > 2000:
return jsonify({
'success': False,
'error': 'Rekomendacja nie może przekraczać 2000 znaków'
}), 400
# Check if user is verified
if not current_user.is_verified:
return jsonify({
'success': False,
'error': 'Tylko zweryfikowani użytkownicy mogą dodawać rekomendacje'
}), 403
# Verify company exists
company = db.query(Company).filter_by(id=company_id, status='active').first()
if not company:
return jsonify({
'success': False,
'error': 'Firma nie znaleziona'
}), 404
# Prevent self-recommendation
if current_user.company_id and current_user.company_id == company_id:
return jsonify({
'success': False,
'error': 'Nie możesz polecać własnej firmy'
}), 400
# Check for duplicate recommendation (user can only have one recommendation per company)
existing_rec = db.query(CompanyRecommendation).filter_by(
user_id=current_user.id,
company_id=company_id
).first()
if existing_rec:
return jsonify({
'success': False,
'error': 'Już poleciłeś tę firmę. Możesz edytować swoją istniejącą rekomendację.'
}), 400
# Create recommendation
recommendation = CompanyRecommendation(
company_id=company_id,
user_id=current_user.id,
recommendation_text=recommendation_text,
service_category=service_category,
show_contact=show_contact,
status='pending' # Start as pending for moderation
)
db.add(recommendation)
db.commit()
db.refresh(recommendation)
# Create notification for company owner (if exists)
# Find users associated with this company
company_users = db.query(User).filter_by(company_id=company_id, is_active=True).all()
for company_user in company_users:
if company_user.id != current_user.id:
notification = UserNotification(
user_id=company_user.id,
notification_type='new_recommendation',
title='Nowa rekomendacja',
message=f'{current_user.name or current_user.email} polecił Twoją firmę: {company.name}',
action_url=f'/company/{company.slug}#recommendations',
related_id=recommendation.id
)
db.add(notification)
db.commit()
logger.info(f"Recommendation created: user {current_user.id} -> company {company_id}, ID {recommendation.id}")
return jsonify({
'success': True,
'message': 'Rekomendacja została utworzona i oczekuje na moderację',
'recommendation_id': recommendation.id,
'status': recommendation.status
}), 201
except Exception as e:
logger.error(f"Error creating recommendation: {e}")
db.rollback()
return jsonify({
'success': False,
'error': 'Wystąpił błąd podczas tworzenia rekomendacji'
}), 500
finally:
db.close()
@app.route('/api/recommendations/<int:rec_id>/edit', methods=['POST'])
@login_required
def api_edit_recommendation(rec_id):
"""API: Edit an existing recommendation (owner or admin only)"""
db = SessionLocal()
try:
# Get the recommendation
recommendation = db.query(CompanyRecommendation).filter_by(id=rec_id).first()
if not recommendation:
return jsonify({
'success': False,
'error': 'Rekomendacja nie znaleziona'
}), 404
# Check authorization - user must be the owner OR admin
if recommendation.user_id != current_user.id and not current_user.is_admin:
return jsonify({
'success': False,
'error': 'Brak uprawnień do edycji tej rekomendacji'
}), 403
# Get JSON data
data = request.get_json()
if not data:
return jsonify({
'success': False,
'error': 'Brak danych'
}), 400
recommendation_text = data.get('recommendation_text', '').strip()
service_category = data.get('service_category', '').strip() or None
show_contact = data.get('show_contact', recommendation.show_contact)
# Validate text if provided
if recommendation_text:
# Validate text length (50-2000 characters)
if len(recommendation_text) < 50:
return jsonify({
'success': False,
'error': 'Rekomendacja musi mieć co najmniej 50 znaków'
}), 400
if len(recommendation_text) > 2000:
return jsonify({
'success': False,
'error': 'Rekomendacja nie może przekraczać 2000 znaków'
}), 400
recommendation.recommendation_text = recommendation_text
# Update other fields if provided
if 'service_category' in data:
recommendation.service_category = service_category
if 'show_contact' in data:
recommendation.show_contact = show_contact
# Update timestamp
recommendation.updated_at = datetime.now()
db.commit()
logger.info(f"Recommendation edited: ID {rec_id} by user {current_user.id}")
return jsonify({
'success': True,
'message': 'Rekomendacja została zaktualizowana',
'recommendation_id': recommendation.id
})
except Exception as e:
logger.error(f"Error editing recommendation {rec_id}: {e}")
db.rollback()
return jsonify({
'success': False,
'error': 'Wystąpił błąd podczas edycji rekomendacji'
}), 500
finally:
db.close()
@app.route('/api/recommendations/<int:rec_id>/delete', methods=['POST'])
@login_required
def api_delete_recommendation(rec_id):
"""API: Delete a recommendation (owner or admin only)"""
db = SessionLocal()
try:
# Get the recommendation
recommendation = db.query(CompanyRecommendation).filter_by(id=rec_id).first()
if not recommendation:
return jsonify({
'success': False,
'error': 'Rekomendacja nie znaleziona'
}), 404
# Check authorization - user must be the owner OR admin
if recommendation.user_id != current_user.id and not current_user.is_admin:
return jsonify({
'success': False,
'error': 'Brak uprawnień do usunięcia tej rekomendacji'
}), 403
# Store info for logging
company_id = recommendation.company_id
user_id = recommendation.user_id
# Delete the recommendation
db.delete(recommendation)
db.commit()
logger.info(f"Recommendation deleted: ID {rec_id} (company {company_id}, user {user_id}) by user {current_user.id}")
return jsonify({
'success': True,
'message': 'Rekomendacja została usunięta'
})
except Exception as e:
logger.error(f"Error deleting recommendation {rec_id}: {e}")
db.rollback()
return jsonify({
'success': False,
'error': 'Wystąpił błąd podczas usuwania rekomendacji'
}), 500
finally:
db.close()
# Routes: /api/recommendations/<company_id>, /api/recommendations/create,
# /api/recommendations/<rec_id>/edit, /api/recommendations/<rec_id>/delete
# ============================================================

1
blueprints/api/__init__.py
View File

@ -10,3 +10,4 @@ from flask import Blueprint
bp = Blueprint('api', __name__, url_prefix='/api')
from . import routes_analytics # noqa: E402, F401
from . import routes_recommendations # noqa: E402, F401

346
blueprints/api/routes_recommendations.py Normal file
View File

@ -0,0 +1,346 @@
"""
Recommendations API Routes - API blueprint
Migrated from app.py as part of the blueprint refactoring.
Contains API routes for company recommendations management.
"""
import logging
from datetime import datetime
from flask import jsonify, request
from flask_login import current_user, login_required
from database import (
SessionLocal,
Company,
User,
CompanyRecommendation,
UserNotification
)
from . import bp
logger = logging.getLogger(__name__)
# ============================================================
# RECOMMENDATIONS API ROUTES
# ============================================================
@bp.route('/recommendations/<int:company_id>', methods=['GET'])
@login_required
def api_get_recommendations(company_id):
"""API: Get all approved recommendations for a company"""
db = SessionLocal()
try:
# Verify company exists
company = db.query(Company).filter_by(id=company_id).first()
if not company:
return jsonify({
'success': False,
'error': 'Firma nie znaleziona'
}), 404
# Query recommendations with user details
recommendations = db.query(CompanyRecommendation).filter_by(
company_id=company_id,
status='approved'
).join(User, CompanyRecommendation.user_id == User.id).order_by(CompanyRecommendation.created_at.desc()).all()
# Build response with recommender details
result = []
for rec in recommendations:
recommender = db.query(User).filter_by(id=rec.user_id).first()
recommender_company = None
if recommender and recommender.company_id:
recommender_company = db.query(Company).filter_by(id=recommender.company_id).first()
rec_data = {
'id': rec.id,
'recommendation_text': rec.recommendation_text,
'service_category': rec.service_category,
'created_at': rec.created_at.isoformat() if rec.created_at else None,
'updated_at': rec.updated_at.isoformat() if rec.updated_at else None,
'recommender': {
'name': recommender.full_name if recommender else '[Użytkownik usunięty]',
'email': recommender.email if (recommender and rec.show_contact) else None,
'phone': recommender.phone if (recommender and rec.show_contact) else None,
'company_id': recommender_company.id if recommender_company else None,
'company_name': recommender_company.name if recommender_company else None,
'company_slug': recommender_company.slug if recommender_company else None
}
}
result.append(rec_data)
return jsonify({
'success': True,
'company_id': company_id,
'company_name': company.name,
'recommendations': result,
'count': len(result)
})
except Exception as e:
logger.error(f"Error fetching recommendations for company {company_id}: {e}")
return jsonify({
'success': False,
'error': 'Wystąpił błąd podczas pobierania rekomendacji'
}), 500
finally:
db.close()
@bp.route('/recommendations/create', methods=['POST'])
@login_required
def api_create_recommendation():
"""API: Create a new recommendation"""
db = SessionLocal()
try:
# Get JSON data
data = request.get_json()
if not data:
return jsonify({
'success': False,
'error': 'Brak danych'
}), 400
company_id = data.get('company_id')
recommendation_text = data.get('recommendation_text', '').strip()
service_category = data.get('service_category', '').strip() or None
show_contact = data.get('show_contact', True)
# Validate required fields
if not company_id:
return jsonify({
'success': False,
'error': 'Brak ID firmy'
}), 400
if not recommendation_text:
return jsonify({
'success': False,
'error': 'Treść rekomendacji jest wymagana'
}), 400
# Validate text length (50-2000 characters)
if len(recommendation_text) < 50:
return jsonify({
'success': False,
'error': 'Rekomendacja musi mieć co najmniej 50 znaków'
}), 400
if len(recommendation_text) > 2000:
return jsonify({
'success': False,
'error': 'Rekomendacja nie może przekraczać 2000 znaków'
}), 400
# Check if user is verified
if not current_user.is_verified:
return jsonify({
'success': False,
'error': 'Tylko zweryfikowani użytkownicy mogą dodawać rekomendacje'
}), 403
# Verify company exists
company = db.query(Company).filter_by(id=company_id, status='active').first()
if not company:
return jsonify({
'success': False,
'error': 'Firma nie znaleziona'
}), 404
# Prevent self-recommendation
if current_user.company_id and current_user.company_id == company_id:
return jsonify({
'success': False,
'error': 'Nie możesz polecać własnej firmy'
}), 400
# Check for duplicate recommendation (user can only have one recommendation per company)
existing_rec = db.query(CompanyRecommendation).filter_by(
user_id=current_user.id,
company_id=company_id
).first()
if existing_rec:
return jsonify({
'success': False,
'error': 'Już poleciłeś tę firmę. Możesz edytować swoją istniejącą rekomendację.'
}), 400
# Create recommendation
recommendation = CompanyRecommendation(
company_id=company_id,
user_id=current_user.id,
recommendation_text=recommendation_text,
service_category=service_category,
show_contact=show_contact,
status='pending' # Start as pending for moderation
)
db.add(recommendation)
db.commit()
db.refresh(recommendation)
# Create notification for company owner (if exists)
# Find users associated with this company
company_users = db.query(User).filter_by(company_id=company_id, is_active=True).all()
for company_user in company_users:
if company_user.id != current_user.id:
notification = UserNotification(
user_id=company_user.id,
notification_type='new_recommendation',
title='Nowa rekomendacja',
message=f'{current_user.name or current_user.email} polecił Twoją firmę: {company.name}',
action_url=f'/company/{company.slug}#recommendations',
related_id=recommendation.id
)
db.add(notification)
db.commit()
logger.info(f"Recommendation created: user {current_user.id} -> company {company_id}, ID {recommendation.id}")
return jsonify({
'success': True,
'message': 'Rekomendacja została utworzona i oczekuje na moderację',
'recommendation_id': recommendation.id,
'status': recommendation.status
}), 201
except Exception as e:
logger.error(f"Error creating recommendation: {e}")
db.rollback()
return jsonify({
'success': False,
'error': 'Wystąpił błąd podczas tworzenia rekomendacji'
}), 500
finally:
db.close()
@bp.route('/recommendations/<int:rec_id>/edit', methods=['POST'])
@login_required
def api_edit_recommendation(rec_id):
"""API: Edit an existing recommendation (owner or admin only)"""
db = SessionLocal()
try:
# Get the recommendation
recommendation = db.query(CompanyRecommendation).filter_by(id=rec_id).first()
if not recommendation:
return jsonify({
'success': False,
'error': 'Rekomendacja nie znaleziona'
}), 404
# Check authorization - user must be the owner OR admin
if recommendation.user_id != current_user.id and not current_user.is_admin:
return jsonify({
'success': False,
'error': 'Brak uprawnień do edycji tej rekomendacji'
}), 403
# Get JSON data
data = request.get_json()
if not data:
return jsonify({
'success': False,
'error': 'Brak danych'
}), 400
recommendation_text = data.get('recommendation_text', '').strip()
service_category = data.get('service_category', '').strip() or None
show_contact = data.get('show_contact', recommendation.show_contact)
# Validate text if provided
if recommendation_text:
# Validate text length (50-2000 characters)
if len(recommendation_text) < 50:
return jsonify({
'success': False,
'error': 'Rekomendacja musi mieć co najmniej 50 znaków'
}), 400
if len(recommendation_text) > 2000:
return jsonify({
'success': False,
'error': 'Rekomendacja nie może przekraczać 2000 znaków'
}), 400
recommendation.recommendation_text = recommendation_text
# Update other fields if provided
if 'service_category' in data:
recommendation.service_category = service_category
if 'show_contact' in data:
recommendation.show_contact = show_contact
# Update timestamp
recommendation.updated_at = datetime.now()
db.commit()
logger.info(f"Recommendation edited: ID {rec_id} by user {current_user.id}")
return jsonify({
'success': True,
'message': 'Rekomendacja została zaktualizowana',
'recommendation_id': recommendation.id
})
except Exception as e:
logger.error(f"Error editing recommendation {rec_id}: {e}")
db.rollback()
return jsonify({
'success': False,
'error': 'Wystąpił błąd podczas edycji rekomendacji'
}), 500
finally:
db.close()
@bp.route('/recommendations/<int:rec_id>/delete', methods=['POST'])
@login_required
def api_delete_recommendation(rec_id):
"""API: Delete a recommendation (owner or admin only)"""
db = SessionLocal()
try:
# Get the recommendation
recommendation = db.query(CompanyRecommendation).filter_by(id=rec_id).first()
if not recommendation:
return jsonify({
'success': False,
'error': 'Rekomendacja nie znaleziona'
}), 404
# Check authorization - user must be the owner OR admin
if recommendation.user_id != current_user.id and not current_user.is_admin:
return jsonify({
'success': False,
'error': 'Brak uprawnień do usunięcia tej rekomendacji'
}), 403
# Store info for logging
company_id = recommendation.company_id
user_id = recommendation.user_id
# Delete the recommendation
db.delete(recommendation)
db.commit()
logger.info(f"Recommendation deleted: ID {rec_id} (company {company_id}, user {user_id}) by user {current_user.id}")
return jsonify({
'success': True,
'message': 'Rekomendacja została usunięta'
})
except Exception as e:
logger.error(f"Error deleting recommendation {rec_id}: {e}")
db.rollback()
return jsonify({
'success': False,
'error': 'Wystąpił błąd podczas usuwania rekomendacji'
}), 500
finally:
db.close()