首页 文章 精选 留言 我的

精选列表

搜索[Python],共10000篇文章
优秀的个人博客,低调大师

Python 爬虫实战:抓取汽车之家二手车成交价格与里程数据

## 1.为什么抓二手车成交数据 二手车是典型的非标品:同一款车,因为上牌年份、行驶里程、所在城市、车况等级不同,价格能差出好几万。汽车之家二手车(che168.com)聚合了全国大量真实成交记录,是做**残值率分析、定价模型、区域行情监控**最省钱的公开数据源。 我们自己要做定价参考时,最关心的两个字段就是: - **成交价格(dealPrice)**:车主实际卖出的价格,比挂牌价更有参考价值; - **行驶里程(mileage)**:衡量车辆损耗的核心指标,通常和价格呈强负相关。 把这两个字段加上车型、城市、上牌时间一起抓下来,就能画出「某车型 / 某市 / 某年款」的价格—里程曲线。举个实际场景:你想买一辆三年的丰田凯美瑞,挂牌价从 11 万到 14 万都有,到底哪个算捡漏?把本地近半年的真实成交记录拉出来,按里程分组算中位数,再对照目标车的里程,一眼就知道报价合不合理。这种「用真实成交价给挂牌价做基准」的用法,比任何估值工具都更贴地气。 更重要的是,成交数据是**时间序列**的——同一款车这个月成交价 12.3 万,下个月 11.8 万,背后可能是新车降价、排放标准切换或季节因素影响。持续抓取、积累历史,你手里就有了一张动态行情表,而不是某个时间点的快照。这正是爬虫相比「手动查一次」最大的价值:它把一次性动作变成了可复用的数据流。 ## 2. 接口分析:别爬 HTML,抓 JSON 接口 直接用 `requests` 把成交列表页拉下来再解析 HTML 是最笨的做法——页面结构一改就全废,而且接口字段往往藏得深。正确姿势是**监听接口**: 1. 打开 che168.com 二手车成交页,按 F12 进 DevTools; 1. 切到 Network 面板,勾选 `XHR / Fetch`; 1. 刷新页面,找返回 `application/json` 的请求,通常带 `getList`、`chengjiao`、`dealList` 之类的关键字; 1. 看它的 Query 参数(page、pageSize、brandId、cityId、seriesId)和 Response 结构(一个 `data.list` 数组)。 拿到真实接口后,直接请求接口比解析 DOM 稳定一个数量级。下面假设我们定位到这样一个接口(实际路径以你抓到为准): ``` GET https://www.che168.com/front/api/usedcar/dealList ?page=1&pageSize=20&cityId=110100&seriesId=0 ``` 几个调试要点,能省你很多坑: - **先看 Preview 再写代码**:DevTools 里点开 Response 的 Preview,确认 `data.list` 里真的有 `dealPrice`、`mileage` 这些字段,以及分页是否靠 `page` 还是 `pageIndex` 控制; - **Cookie 可能不是必须的**:很多列表接口是公开匿名接口,但带上一两个关键 Cookie(如 `fvlid`、`sessionid`)会更像真人,降低被拦概率; - **参数别多带**:只保留翻页和筛选相关的参数,多余的参数可能是签名校验来源,少带反而更干净; - **注意 gzip**:`requests` 默认会自动解压,但如果自己用 `httpx` 或 `aiohttp`,记得开启 `follow_redirects` 并处理好压缩,否则拿到的是乱码。 ## 3. 最小可运行爬虫 先写一个不带代理的版本,确认接口能通: ``` import time import requests BASE_URL = "https://www.che168.com/front/api/usedcar/dealList" HEADERS = { "User-Agent": ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/124.0 Safari/537.36" ), "Referer": "https://www.che168.com/", "Accept": "application/json, text/plain, */*", } def fetch_deals(city_id: int, max_pages: int = 5) -> list[dict]: """抓取指定城市的二手车成交记录。 Args: city_id: 城市编码,如北京 110100。 max_pages: 最多翻页数,防止无限爬取。 Returns: 成交记录字典列表,每条含车型/价格/里程等字段。 """ results: list[dict] = [] for page in range(1, max_pages + 1): params = {"page": page, "pageSize": 20, "cityId": city_id} resp = requests.get(BASE_URL, params=params, headers=HEADERS, timeout=10) resp.raise_for_status() payload = resp.json() items = payload.get("data", {}).get("list", []) if not items: break results.extend(items) print(f"page={page} got {len(items)} items") time.sleep(1.5) # 礼貌性限速,别把人家打挂 return results if __name__ == "__main__": deals = fetch_deals(city_id=110100) print(f"total: {len(deals)}") ``` 字段映射时留意:成交价通常在 `dealPrice` 或 `price` 里(单位元),里程在 `mileage`(单位万公里)或 `licheng`,上牌时间在 `registerDate` / `cardDate`。**以你实际抓到的 JSON 为准**,不要照抄字段名。 ## 4. 接入云代理 IP 单机直连可用于小规模采集,但批量抓取全国数据时,目标站点会基于出口 IP 做限流或封禁,引入代理 IP 是必要手段。 本文采用**亿牛云(16yun.cn)动态转发代理**:所有请求统一经其转发网关出口,由后台按请求轮换出口 IP,调用方无需自建与维护 IP 池。接入方式即在 `requests` 的 `proxies` 中配置转发网关,认证凭据为订单号(用户名)与订单密码,用户名可按 `-地区-ttl` 追加出口地域与存活时长参数: ``` # 亿牛云动态转发代理(16yun.cn) # 统一转发网关,后台按请求轮换出口 IP,无需自建 IP 池 PROXY_HOST = "t.16yun.cn" PROXY_PORT = "31111" PROXY_USER = "YOUR_ORDER_ID" # 亿牛云订单号 PROXY_PASS = "YOUR_PASSWORD" # 订单密码 PROXIES = { "http": f"http://{PROXY_USER}:{PROXY_PASS}@{PROXY_HOST}:{PROXY_PORT}", "https": f"http://{PROXY_USER}:{PROXY_PASS}@{PROXY_HOST}:{PROXY_PORT}", } ``` 在请求中挂载代理: ``` resp = requests.get( BASE_URL, params=params, headers=HEADERS, proxies=PROXIES, timeout=10, ) ``` 工程上建议将代理请求封装为 `get_with_proxy()`,内部对超时/连接异常做有限次重试(如 3 次),以吸收单点代理抖动。亿牛云另提供 API 提取模式(主动拉取 IP 池自行调度),适用于需要精细控制并发与地域分布的场景;动态转发模式运维成本更低,更契合本文的中等规模采集。 ## 5. 数据清洗与落库 抓回来的原始 JSON 还不能直接分析。里程可能是 `"3.2万公里"` 这种字符串,价格可能带单位。用 Pandas 统一清洗: ``` import pandas as pd import re def clean_deals(raw: list[dict]) -> pd.DataFrame: """把原始成交记录清洗成规整的 DataFrame。 Args: raw: 接口返回的原始记录列表。 Returns: 规整后的 DataFrame,价格/里程均为 float。 """ rows = [] for it in raw: price = it.get("dealPrice") or it.get("price") mileage = it.get("mileage") or it.get("licheng") rows.append({ "model": it.get("seriesName") or it.get("carName"), "city": it.get("cityName"), "register_date": it.get("registerDate"), "deal_price": float(re.sub(r"[^\d.]", "", str(price))) if price else None, "mileage": float(re.sub(r"[^\d.]", "", str(mileage))) if mileage else None, }) df = pd.DataFrame(rows).drop_duplicates() return df if __name__ == "__main__": deals = fetch_deals(city_id=110100) df = clean_deals(deals) df.to_csv("used_car_deals.csv", index=False, encoding="utf-8-sig") print(df.describe()) # 快速看价格/里程分布 ``` `encoding="utf-8-sig"` 是为了让 Excel 打开中文不乱码。如果需要长期追踪价格变动,把 `df` 换成 upsert 写入 MySQL(用唯一键 `车型+城市+上牌时间`)即可建立历史价格表。 ## 6. 拿到数据能做什么:价格—里程相关性分析 数据落库后,最有代表性的一个分析就是看「里程越高、价格越低」这件事到底有多强。用 `matplotlib` 画个散点图加趋势线: ``` import matplotlib.pyplot as plt import numpy as np def plot_price_vs_mileage(df: pd.DataFrame, model: str) -> None: """绘制指定车型的价格—里程散点图与趋势线。 Args: df: 清洗后的成交 DataFrame。 model: 车型名,用于筛选与标题。 """ sub = df[df["model"] == model].dropna(subset=["deal_price", "mileage"]) if len(sub) pd.DataFrame: """采集单个城市并清洗,返回 DataFrame。""" return clean_deals(fetch_deals(city_id, max_pages=10)) with ThreadPoolExecutor(max_workers=4) as pool: frames = list(pool.map(crawl_city, CITY_IDS)) df_all = pd.concat(frames, ignore_index=True) df_all.to_csv("used_car_deals_multi.csv", index=False, encoding="utf-8-sig") ``` 并发数别贪多,`max_workers` 控制在 4~8 之间,配合每请求 1 秒左右的限速,对目标站点足够友好,也足够把你一天的量在几分钟内跑完。 ## 8. 反爬与工程化注意事项 1. **限速优先**:`time.sleep` 别省,配合亿牛云轮换 IP,单机也能稳稳爬几千条; 1. **Header 完整**:`Referer`、`User-Agent` 必须带,`Accept` 标明 json,否则接口可能返回 403; 1. **UA 轮换**:准备一个 UA 池随机挑,避免单个 UA 特征被识别; 1. **签名/Token**:部分接口会校验 `sign` 或 `token`,需要逆向 JS 还原算法(这是进阶关卡,本文先不展开); 1. **字体反爬**:极少数页面用自定义字体混淆数字,需要下载字体文件做映射还原; 1. **异常兜底**:网络超时、JSON 解析失败都要 `try/except` 包住,单条失败不影响整体; 1. **合规边界**:仅抓取公开数据用于个人分析,控制频率,遵守 `robots.txt`,不碰任何隐私字段。 额外提醒一个工程化细节:**断点续爬**。批量采集经常因为网络抖动中断,建议在落盘时按 `city_id + page` 记录进度,重启后跳过已抓的页,而不是从头再来。可以用一个本地 `set` 或 SQLite 小表存「已完成的 page 键」,每次抓取前先查一下。 还有一个容易被忽略的数据质量问题:成交记录里偶尔会出现「价格 0」「里程 0」或明显离谱的异常值(比如标价误填成成交价)。清洗阶段一定要加一道规则过滤——比如 `deal_price` 低于 0.5 万或高于 500 万的直接丢弃,`mileage` 超过 50 万公里的视为脏数据。宁可少几条,也别让一个离群点把整条趋势线带歪。 ## 9. 小结 这条链路的核心就三步:**抓接口不抓页面 → 用亿牛云代理绕开限流 → Pandas 清洗落库**。接口逆向让爬虫更稳,动态代理让爬虫更久,清洗落库让数据真正可用。把成交价和里程抓全之后,你就能做任何想做的分析了——比如算某款车的「每万公里贬值曲线」,或者对比一线城市和二线城市的同车价差。

优秀的个人博客,低调大师

用 Python 做了一个AI日报生成器,每天自动汇总行业动态

![](https://developer.qcloudimg.com/http-save/yehe-11081884/46d54d0bd585a61be2d334caf609a874.jpg) 一家做供应链SaaS的创业公司,二十来个人。老板姓李,做销售出身,对行业信息特别敏感。 有天早会他突然说:“从明天开始,每天早上九点之前,往群里发一份行业简报。把物流、供应链、SaaS这几个领域的最新动态整理一下,大家上班就能看。” 每天早上九点之前发,意味着八点多就得起来干活。手动去翻十几个网站,挑新闻,写摘要,至少得一个小时。 于是开始琢磨:这事必须自动化。 --- 整个工具分三步。第一步,用requests爬几个目标网站的新闻列表,提取标题和摘要。第二步,把抓到的新闻丢给大模型,让它做分类、去重、总结,生成一份有结构的简报。第三步,用定时任务每天自动跑,生成的简报存成文件或者直接发邮件。 爬虫部分我没有用Selenium那种重型方案,因为大多数新闻站的列表页都是服务端渲染的,requests加BeautifulSoup就够了。 --- 先装依赖: ``` pip install requests beautifulsoup4 schedule openpyxl ``` 完整代码: ``` import requests from bs4 import BeautifulSoup import json import time import os import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from datetime import datetime # ============ 配置区域 ============ LLM_API_KEY = "sk-your-deepseek-api-key" LLM_API_URL = "https://api.deepseek.com/chat/completions" # 邮件发送配置(可选,不发邮件也行) EMAIL_ENABLED = False SMTP_SERVER = "smtp.qq.com" SMTP_PORT = 465 EMAIL_FROM = "your-email@qq.com" EMAIL_PASSWORD = "your-smtp-password" EMAIL_TO = ["boss@example.com", "team@example.com"] # 新闻源配置:网站名称和对应的爬取规则 NEWS_SOURCES = [ { "name": "36氪", "url": "https://36kr.com/newsflashes", "parser": "parse_36kr" }, { "name": "物流沙龙", "url": "https://www.logclub.com/news", "parser": "parse_generic" }, { "name": "亿邦动力", "url": "https://www.ebrun.com/news/", "parser": "parse_generic" } ] # 通用请求头,模拟浏览器 HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8" } # ============ 第一步:新闻爬取 ============ def fetch_page(url, timeout=15): """ 请求网页并返回BeautifulSoup对象 加入重试机制和错误处理 """ try: response = requests.get(url, headers=HEADERS, timeout=timeout) response.raise_for_status() # 自动检测编码 response.encoding = response.apparent_encoding soup = BeautifulSoup(response.text, "html.parser") return soup except requests.RequestExceptionase: print(f" [请求失败] {url}: {e}") return None def parse_36kr(url): """ 解析36氪快讯页面 提取快讯标题和简要内容 """ soup = fetch_page(url) if not soup: return [] news_list = [] # 36氪快讯的条目(页面结构可能变化,需根据实际调整选择器) items = soup.select(".newsflash-item") orsoup.select("[class*='flash']") orsoup.select("article") for item in items[:20]: # 取前20条 title_elem = item.select_one("a") or item.select_one("h3") or item.select_one(".title") if not title_elem: continue title = title_elem.get_text(strip=True) link = title_elem.get("href", "") if link and not link.startswith("http"): link = "https://36kr.com"+link # 尝试获取摘要 desc_elem = item.select_one(".description") or item.select_one("p") or item.select_one(".content") desc = desc_elem.get_text(strip=True) if desc_elem else"" news_list.append({ "title": title, "description": desc[:200], # 摘要截断到200字 "url": link, "source": "36氪" }) return news_list def parse_generic(url): """ 通用的新闻列表解析器 适用于大多数新闻网站的文章列表页 """ soup = fetch_page(url) if not soup: return [] news_list = [] # 尝试多种常见的文章列表选择器 articles = ( soup.select("article") or soup.select(".news-item") or soup.select(".article-item") or soup.select(".post-item") or soup.select("li.item") or soup.select(".list-item") ) for article in articles[:20]: # 提取标题 title_elem = ( article.select_one("h2 a") or article.select_one("h3 a") or article.select_one(".title a") or article.select_one("a") ) if not title_elem: continue title = title_elem.get_text(strip=True) if len(title) ") html_body = f"{html_body}" msg.attach(MIMEText(body, "plain", "utf-8")) msg.attach(MIMEText(html_body, "html", "utf-8")) try: with smtplib.SMTP_SSL(SMTP_SERVER, SMTP_PORT) asserver: server.login(EMAIL_FROM, EMAIL_PASSWORD) server.sendmail(EMAIL_FROM, EMAIL_TO, msg.as_string()) print(f"邮件已发送到:{', '.join(EMAIL_TO)}") except Exception as e: print(f"邮件发送失败:{e}") # ============ 定时任务 ============ def run_daily_job(): """ 完整的日报生成流程:爬取 -> 摘要 -> 保存 -> 发送 """ print(f"\n{'='*50}") print(f"行业日报生成器 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"{'='*50}") # 1. 爬取新闻 print("\n[1/4] 开始爬取新闻...") news_list = crawl_all_sources() print(f"共抓取到 {len(news_list)} 条新闻") if not news_list: print("未抓取到任何新闻,请检查网络和新闻源配置") return # 2. AI生成日报 print("\n[2/4] 正在调用AI生成日报...") report = generate_daily_report(news_list) # 3. 保存文件 print("\n[3/4] 保存日报文件...") filepath = save_report(report) # 4. 发送邮件 print("\n[4/4] 发送邮件...") date_str = datetime.now().strftime("%m月%d日") send_email(f"行业日报 - {date_str}", report) print(f"\n{'='*50}") print("日报生成完毕") print(f"{'='*50}\n") def schedule_daily_job(): """ 设置定时任务,每天早上8:30自动执行 (8:30跑,9:00之前肯定能生成完) """ import schedule # 每天早上8:30执行 schedule.every().day.at("08:30").do(run_daily_job) print("定时任务已启动") print("每天 08:30 自动生成行业日报") print("按 Ctrl+C 停止\n") while True: schedule.run_pending() time.sleep(30) # ============ 主入口 ============ if __name__ == "__main__": import sys if len(sys.argv) >1 and sys.argv[1] == "--schedule": # 定时模式 schedule_daily_job() else: # 单次运行模式(测试用) run_daily_job() ``` --- 运行效果: ``` ================================================== 行业日报生成器 - 2024-12-20 08:30:05 ================================================== [1/4] 开始爬取新闻... 正在爬取:36氪 (https://36kr.com/newsflashes) 获取到 18 条新闻 正在爬取:物流沙龙 (https://www.logclub.com/news) 获取到 12 条新闻 正在爬取:亿邦动力 (https://www.ebrun.com/news/) 获取到 15 条新闻 共抓取到 45 条新闻 [2/4] 正在调用AI生成日报... [3/4] 保存日报文件... 日报已保存到:./reports/industry_report_20241220.md [4/4] 发送邮件... 邮件发送未启用,跳过 ================================================== 日报生成完毕 ================================================== ``` 生成的日报大概长这样: ``` # 行业日报 - 2024年12月20日 ## 今日要点 京东物流宣布将在2025年新增50个智能仓储中心,总投资超过80亿元,进一步扩大其供应链基础设施布局。与此同时,SaaS赛道迎来一笔大额融资,供应链管理平台"链易"完成B轮2亿元融资。政策层面,交通运输部发布了新版《网络货运经营管理暂行办法》,对平台型物流企业的合规要求做了细化。 ## 物流与供应链 **京东物流2025年新增50个智能仓储中心** 京东物流披露明年基建计划,80亿投资规模创近三年新高。这释放出一个信号:头部物流企业在降本增效的同时,仍在加码重资产投入,行业集中度会进一步提升。 **菜鸟发布"全球5日达"升级计划** 菜鸟国际快递宣布跨境时效覆盖扩至全球50个核心城市。跨境电商卖家将是最大受益者,物流时效直接影响复购率。 **交通运输部发布网络货运新规** 新规要求平台企业对实际承运人资质做实质性审核,不再是"挂个名就行"。中小网络货运平台合规成本会上升,行业可能迎来一波洗牌。 ## SaaS与企业服务 **供应链SaaS"链易"完成B轮2亿融资** 链易主打中小制造企业的采购协同,这轮融资由红杉领投。说明资本对垂直SaaS仍有兴趣,关键看能不能真正解决行业痛点。 **钉钉发布制造业专属解决方案** 钉钉针对制造业场景推出了生产管理、质量追溯等模块。大厂在垂直行业做深,中小SaaS厂商的差异化空间被压缩了。 ## 编辑观点 今天最值得关注的信号是物流行业的"两头分化"。一边是京东、菜鸟这样的巨头在基础设施上持续加码,一边是监管对中小平台合规要求收紧。对SaaS创业者来说,机会可能在于:巨头做标准化的重资产,你可以做它们不愿意碰的定制化轻服务,嵌进巨头的生态里活下来。 ``` --- 看到这个日报之后非常满意。AI每次都会给一个“编辑观点”,有点分析的味道,不像是简单搬运。 后来又加了个功能:把日报自动发到企业微信群里。用的是企业微信的群机器人webhook,一行POST请求就行,比发邮件还简单。从那以后,每天九点上班打开电脑,日报已经躺在那了。 `NEWS_SOURCES` 那个列表你可以随便改,加上你关注的网站。如果某个网站的结构比较特殊,照着`parse_36kr`那个函数写一个专门的解析器就行。爬虫这东西,最大的成本就是维护,网站改版了你得跟着改选择器。不过好在AI摘要那部分不用动,新闻格式怎么变,AI都能处理。 **“无他,惟手熟尔”!**有需要的用起来!

资源下载

更多资源
腾讯云软件源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题,腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构,目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称,一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集,帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Rocky Linux

Rocky Linux

Rocky Linux(中文名:洛基)是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版,作为CentOS稳定版停止维护后与RHEL(Red Hat Enterprise Linux)完全兼容的开源替代方案,由社区拥有并管理,支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性,采用模块化包装和SELinux安全架构,默认包含GNOME桌面环境及XFS文件系统,支持十年生命周期更新。

Sublime Text

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能,例如代码缩略图,Python的插件,代码段等。还可自定义键绑定,菜单和工具栏。Sublime Text 的主要功能包括:拼写检查,书签,完整的 Python API , Goto 功能,即时项目切换,多选择,多窗口等等。Sublime Text 是一个跨平台的编辑器,同时支持Windows、Linux、Mac OS X等操作系统。

用户登录
用户注册