人工智能之编程进阶 Python高级：第十一章 过渡项目

# 示例：通过经纬度查天气（需先查城市坐标）
def get_weather(lat, lon):
    url = f"https://api.open-meteo.com/v1/forecast"
    params = {
        "latitude": lat,
        "longitude": lon,
        "current_weather": True
    }
    resp = requests.get(url, params=params)
    data = resp.json()
    return data["current_weather"]

import argparse

parser = argparse.ArgumentParser(description="天气查询工具")
parser.add_argument("city", help="城市名，如 Beijing")
args = parser.parse_args()

weather = get_weather_by_city(args.city)
print(f"{args.city} 当前温度: {weather['temperature']}°C")

CITY_COORDS = {
    "Beijing": (39.9042, 116.4074),
    "Shanghai": (31.2304, 121.4737),
    # 可扩展为读取 CSV 或调用地理编码 API
}

url = "https://movie.douban.com/top250"
headers = {"User-Agent": "Mozilla/5.0"}
resp = requests.get(url, headers=headers)
soup = BeautifulSoup(resp.text, "lxml")

for item in soup.select(".item"):
    title = item.select_one(".title").text
    rating = item.select_one(".rating_num").text
    year = item.select_one(".bd p").text.split()[-1].strip("()")

all_movies = []
for start in range(0, 250, 25):
    page_url = f"https://movie.douban.com/top250?start={start}"
    movies = parse_page(page_url)
    all_movies.extend(movies)
    time.sleep(1)  # 礼貌延迟

import pandas as pd

df = pd.DataFrame(all_movies)
df.to_csv("douban_top250.csv", index=False)

print("平均分:", df["rating"].astype(float).mean())
print("最高分电影:\n", df.loc[df["rating"].astype(float).idxmax()])

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

options = webdriver.ChromeOptions()
options.add_argument("--headless")
driver = webdriver.Chrome(options=options)

driver.get("https://github.com/login")
driver.find_element(By.ID, "login_field").send_keys("your_email")
driver.find_element(By.ID, "password").send_keys("your_password")
driver.find_element(By.NAME, "commit").click()

# 等待登录成功
WebDriverWait(driver, 10).until(
    EC.presence_of_element_located((By.CSS_SELECTOR, "header[role='banner']"))
)

driver.get(f"https://github.com/{username}?tab=repositories")
repos = []
for repo in driver.find_elements(By.CSS_SELECTOR, "li.public"):
    name = repo.find_element(By.CSS_SELECTOR, "h3 a").text
    stars = repo.find_element(By.CSS_SELECTOR, "svg.octicon-star").text.strip()
    repos.append({"name": name, "stars": stars})

import aiohttp
import asyncio

async def fetch(session, url):
    try:
        async with session.get(url, timeout=10) as resp:
            return await resp.text()
    except Exception as e:
        return None

async def main():
    urls = [line.strip() for line in open("urls.txt")]
    
    # 限制并发数（防止被封）
    semaphore = asyncio.Semaphore(20)
    
    async def bounded_fetch(url):
        async with semaphore:
            return await fetch(session, url)
    
    async with aiohttp.ClientSession() as session:
        tasks = [bounded_fetch(url) for url in urls]
        results = await asyncio.gather(*tasks)
    
    print(f"成功: {len([r for r in results if r])} / {len(urls)}")

class NewsSpider(scrapy.Spider):
    name = "news"
    start_urls = [
        "https://www.bbc.com/news",
        "https://www.reuters.com/world/"
    ]

    def parse(self, response):
        if "bbc.com" in response.url:
            yield from self.parse_bbc(response)
        elif "reuters.com" in response.url:
            yield from self.parse_reuters(response)

# pipelines.py
class PostgresPipeline:
    def open_spider(self, spider):
        self.conn = psycopg2.connect(...)
        self.cur = self.conn.cursor()

    def process_item(self, item, spider):
        self.cur.execute(
            "INSERT INTO news (title, content, url) VALUES (%s, %s, %s)",
            (item['title'], item['content'], item['url'])
        )
        self.conn.commit()
        return item

# api.py
from fastapi import FastAPI
import psycopg2

app = FastAPI()

@app.get("/news")
def get_news(keyword: str = None):
    # 查询数据库
    cur.execute("SELECT * FROM news WHERE title ILIKE %s", (f"%{keyword}%",))
    return cur.fetchall()

FROM python:3.10
COPY . /app
WORKDIR /app
RUN pip install -r requirements.txt
CMD ["scrapy", "crawl", "news"]