Python脚本开发求助：API调用、JSON解析与循环调用及邮件发送

Hey there! Let's work through this together. I see you're building a Python script that chains API calls, parses JSON data, and sends the final results to your team—totally relatable workflow, but it's easy to hit snags when dealing with nested APIs and JSON parsing. Since you mentioned you've hit a wall even after checking tutorials, let's break down a solid framework for your script, plus point out common pitfalls to look out for.

Core Script Framework #

Here's a structured example that covers every step you're trying to implement—feel free to tweak it to match your actual API endpoints and data structures:

import requests
import json
# Example using email for notifications; swap with Slack/Teams/other tools as needed
import smtplib
from email.mime.text import MIMEText

# Configure your API credentials and endpoints
API_KEY = "your_actual_api_key_here"
INITIAL_API_URL = "https://your-first-api-endpoint.com/fetch-data"
SECOND_API_TEMPLATE = "https://your-second-api-endpoint.com/detail/{target_value}"

# Setup headers (adjust based on your API's auth requirements)
request_headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}

try:
    # Step 1: Fetch initial JSON data
    initial_response = requests.get(INITIAL_API_URL, headers=request_headers)
    # Fail fast if the request didn't succeed
    initial_response.raise_for_status()
    initial_json = initial_response.json()

    # Step 2: Extract specific values from initial JSON
    # Replace this with your actual field path—use print(json.dumps(initial_json, indent=2)) to inspect structure
    target_values = [item["unique_id"] for item in initial_json.get("data", {}).get("entries", [])]

    if not target_values:
        print("No target values extracted! Double-check your JSON structure and field names.")
        exit()

    # Step 3: Loop through values to fetch second set of data
    final_results = []
    for value in target_values:
        # Format the second API URL with the current value
        second_api_url = SECOND_API_TEMPLATE.format(target_value=value)
        
        try:
            second_response = requests.get(second_api_url, headers=request_headers)
            second_response.raise_for_status()
            detail_json = second_response.json()

            # Extract the specific result you need from the detail JSON
            result_entry = {
                "source_id": value,
                "key_result": detail_json.get("critical_field", "No result found")
            }
            final_results.append(result_entry)

        except requests.exceptions.RequestException as err:
            # Catch and log errors for individual requests without breaking the loop
            print(f"Failed to fetch data for {value}: {str(err)}")
            final_results.append({"source_id": value, "key_result": f"Error: {str(err)}"})

    # Step 4: Send results to stakeholders
    def send_notification(results):
        # Build an HTML email (adjust content to your needs)
        email_content = "<h2>API Data Processing Results</h2><ul>"
        for res in results:
            email_content += f"<li>Source ID: {res['source_id']} | Result: {res['key_result']}</li>"
        email_content += "</ul>"

        msg = MIMEText(email_content, "html")
        msg["Subject"] = "Your API Data Update"
        msg["From"] = "your-email@example.com"
        msg["To"] = "team-member@example.com, another-person@example.com"

        # Send via SMTP (update with your email provider's settings)
        with smtplib.SMTP("smtp.example.com", 587) as smtp_server:
            smtp_server.starttls()
            smtp_server.login("your-email@example.com", "your-email-password")
            smtp_server.send_message(msg)
        print("Results sent successfully!")

    send_notification(final_results)

except requests.exceptions.RequestException as init_err:
    print(f"Initial API request failed: {str(init_err)}")
except json.JSONDecodeError as json_err:
    # If JSON parsing fails, print the raw response to debug
    print(f"Failed to parse initial JSON: {str(json_err)}")
    print(f"Raw response snippet: {initial_response.text[:500]}...")
except Exception as general_err:
    print(f"Unexpected error: {str(general_err)}")

Common Troubleshooting Tips #

Since you didn't share the full error output, here are the most likely culprits:

• Auth Issues: Double-check your API key format—some APIs use API-Key: YOUR_KEY instead of Bearer auth. Verify with the API docs.
• JSON Structure Mismatch: Print the formatted initial JSON with print(json.dumps(initial_json, indent=2)) to confirm your target fields exist (they might be nested under data or results instead of top-level).
• API Rate Limits: If some loop requests fail randomly, check if the API has rate limits—you might need to add a time.sleep(1) between requests.
• Non-JSON Responses: If you get a JSON decode error, print the raw response text—you might be getting an HTML error page (like 404 or 500) instead of valid JSON.

If you can share the full error message and a snippet of the initial JSON response (redact sensitive data!), I can help you pinpoint the exact issue.

内容的提问来源于stack exchange，提问作者Kalim