from flask import Flask, render_template, request, jsonify import os import base64 import requests from PIL import Image from io import BytesIO import logging from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry # Configure logging logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s') app = Flask(__name__) # Configuration UPLOAD_FOLDER = 'static/captures' os.makedirs(UPLOAD_FOLDER, exist_ok=True) app.config['UPLOAD_FOLDER'] = UPLOAD_FOLDER # Hugging Face API settings HF_API_TOKEN = os.getenv('HF_API_TOKEN') # Load token from environment variable HF_API_URL = "https://api-inference.huggingface.co/models/facebook/detr-resnet-50" # Example model for object detection # Instagram API settings INSTAGRAM_ACCESS_TOKEN = os.getenv('INSTAGRAM_ACCESS_TOKEN') # Load token from environment variable INSTAGRAM_API_URL = "https://graph.instagram.com/v21.0/me/media" def query_hugging_face(image_data): if not HF_API_TOKEN: logging.error("Hugging Face API token not set.") return {"error": "Hugging Face API token not set. Please set the HF_API_TOKEN environment variable."} try: session = requests.Session() # Configure retries: 3 attempts, backoff factor of 1s, retry on 502/503/504 retries = Retry(total=3, backoff_factor=1, status_forcelist=[502, 503, 504]) session.mount('https://', HTTPAdapter(max_retries=retries)) headers = {"Authorization": f"Bearer {HF_API_TOKEN}"} logging.debug(f"Sending request to {HF_API_URL}") response = session.post(HF_API_URL, headers=headers, data=image_data, timeout=10) response.raise_for_status() logging.debug("Hugging Face API request successful.") return response.json() except requests.exceptions.RequestException as e: logging.error(f"Failed to connect to Hugging Face API: {str(e)}") return {"error": f"Failed to connect to Hugging Face API: {str(e)}"} def upload_to_instagram(image_path): if not INSTAGRAM_ACCESS_TOKEN: logging.error("Instagram access token not set.") return {"error": "Instagram access token not set. Please set the INSTAGRAM_ACCESS_TOKEN environment variable."} try: # Step 1: Create media object session = requests.Session() retries = Retry(total=3, backoff_factor=1, status_forcelist=[502, 503, 504]) session.mount('https://', HTTPAdapter(max_retries=retries)) media_url = f"{INSTAGRAM_API_URL}?image_url={request.url_root}{image_path}&caption=Uploaded%20via%20Flask%20Camera%20App&access_token={INSTAGRAM_ACCESS_TOKEN}" logging.debug(f"Sending media creation request to Instagram: {media_url}") response = session.post(media_url, timeout=10) response.raise_for_status() media_data = response.json() if 'id' not in media_data: logging.error(f"Failed to create media: {media_data}") return {"error": f"Failed to create media: {media_data}"} # Step 2: Publish media media_id = media_data['id'] publish_url = f"https://graph.instagram.com/v21.0/me/media_publish?creation_id={media_id}&access_token={INSTAGRAM_ACCESS_TOKEN}" logging.debug(f"Sending publish request to Instagram: {publish_url}") publish_response = session.post(publish_url, timeout=10) publish_response.raise_for_status() logging.debug("Instagram upload successful.") return {"status": "success"} except requests.exceptions.RequestException as e: logging.error(f"Failed to upload to Instagram: {str(e)}") return {"error": f"Failed to upload to Instagram: {str(e)}"} @app.route('/') def index(): return render_template('index.html') @app.route('/capture', methods=['POST']) def capture(): try: # Get the base64 image data from the request data = request.form['image'] header, encoded = data.split(",", 1) binary_data = base64.b64decode(encoded) # Save the image filename = f"capture_{len(os.listdir(app.config['UPLOAD_FOLDER'])) + 1}.jpg" filepath = os.path.join(app.config['UPLOAD_FOLDER'], filename) with open(filepath, "wb") as f: f.write(binary_data) # Process with Hugging Face API with open(filepath, "rb") as f: hf_result = query_hugging_face(f.read()) # Return the image URL and Hugging Face result image_url = f"/{filepath}" return jsonify({ 'status': 'success', 'image_url': image_url, 'hf_result': hf_result }) except Exception as e: logging.error(f"Error in capture: {str(e)}") return jsonify({'status': 'error', 'message': str(e)}) @app.route('/upload_instagram', methods=['POST']) def upload_instagram(): try: image_url = request.form['image_url'] if not image_url.startswith('/static/captures/'): return jsonify({'status': 'error', 'message': 'Invalid image path.'}) result = upload_to_instagram(image_url.lstrip('/')) if 'error' in result: return jsonify({'status': 'error', 'message': result['error']}) return jsonify({'status': 'success'}) except Exception as e: logging.error(f"Error in upload_instagram: {str(e)}") return jsonify({'status': 'error', 'message': str(e)}) @app.route('/test_connectivity') def test_connectivity(): try: response = requests.get('https://www.google.com', timeout=5) logging.debug(f"Connectivity test to Google: {response.status_code}") return jsonify({'status': 'success', 'code': response.status_code}) except requests.exceptions.RequestException as e: logging.error(f"Connectivity test failed: {str(e)}") return jsonify({'status': 'error', 'message': str(e)}) if __name__ == '__main__': app.run(debug=True, host='0.0.0.0', port=7860)