懒人摄影家：如何用vivo X100和Python打造个人专属摄影展（照片收集篇）
前篇：手机摄影爱好者的自动化（照片收集篇）
idea起源
在繁忙的生活节奏中寻找片刻宁静，记录下那些转瞬即逝的美好瞬间，对于许多人来说，手机摄影成为了最佳的选择。作为一名不抽烟、不喝酒、不打牌的“懒人”，我找到了一种既经济实惠又省时省力的方式来享受摄影——通过我的vivo X100智能手机。
自从有了孩子后，我对任何需要花费大量时间和金钱的兴趣爱好都望而却步。然而，手机摄影不仅满足了我对美的追求，而且几乎不需要额外的投资。特别是当我看到邻居家的宝妈使用X80拍摄出令人惊艳的照片后，我毫不犹豫地升级了我的设备。
入手vivo X100后，我发现了一个名为vivo摄影的应用程序。这个应用不仅让分享和整理照片变得轻松愉快，还让我能够专注于捕捉生活中的每一个精彩瞬间，而不必担心照片杂乱无章的问题。于是，我开始将拍摄的家人以外的作品上传到vivo摄影，并定期清理手机上的本地副本，确保手机存储空间的高效利用。
我的vivo摄影官方展示网址：
https://gallery.vivo.com.cn/galleryh5static/index.html#/user/599660628355
但是，有一个问题一直困扰着我：vivo摄影的在线展示页面底部总是挂着一个显眼的下载横幅，这严重影响了我的观感体验。作为一个有轻微强迫症的人，我渴望拥有一个更加整洁美观的个人摄影作品展示平台。虽然我不擅长编程，但我还是决定尝试一下，看看是否能通过技术手段解决这个问题。
解决方案
经过一番研究，我设计了一套解决方案：通过Python脚本自动收集vivo摄影中的照片URL，并将其保存到数据库中，最终在我的个人网站或微信小程序上展示这些照片。这样，不仅能保持作品集的整洁美观，还能方便快捷地与亲朋好友分享我的创作成果。
具体实现步骤如下：
首先，我们需要编写一个用于管理数据库连接的类。这里我们使用MariaDB作为数据库管理系统，并通过Python来实现：
import mariadb
import logging
class MySQLDatabase:
def __init__(self, host, port, database, user, password,pool_size=5):
self.host = host
self.port = port
self.database = database
self.user = user
self.password = password
self.pool_size = pool_size
self.pool = None
logging.info(f"🔔初始化 MySQLDatabase: host={host}, port={port}, database={database}, user={user}")
def connect(self):
try:
# 先尝试连接到 MySQL 服务器，不指定数据库
self.pool = mariadb.ConnectionPool(
pool_name="db_pool",
pool_size=self.pool_size,
host=self.host,
port=self.port,
user=self.user,
password=self.password
)
logging.info("✅ MySQL 数据库连接池创建成功")
# 检查并创建数据库
self.create_database_if_not_exists()
# 重新连接到指定的数据库
self.pool = mariadb.ConnectionPool(
pool_name="vivo_pool",
pool_size=self.pool_size,
host=self.host,
port=self.port,
database=self.database,
user=self.user,
password=self.password
)
logging.info(f"✅ 连接到数据库 {self.database} 成功")
except mariadb.Error as e:
logging.error(f"❌ 连接失败: {e}")
def create_database_if_not_exists(self):
try:
# 获取连接并创建数据库
conn = self.pool.get_connection()
cursor = conn.cursor()
cursor.execute(f"CREATE DATABASE IF NOT EXISTS {self.database}")
logging.info(f"✅ 数据库 {self.database} 已创建或已存在。")
cursor.close()
conn.close()
except mariadb.Error as err:
logging.error(f"❌ 创建数据库失败: {err}")
def disconnect(self):
if self.pool:
self.pool.close()
logging.info("✅MySQL 数据库连接池已关闭")
def get_connection(self):
try:
return self.pool.get_connection()
except mariadb.Error as e:
logging.error(f"获取连接失败: {e}")
return None
这个类提供了数据库的基本操作功能，如连接、断开连接、以及检查和创建数据库等。接下来，我们将继续开发用于抓取vivo摄影中的照片URL并将它们存储到数据库中的脚本。
该脚本的规则如下：
- 获取照片posts
- 根据每个posts获取下面的image列表
- 将post的信息和image的url写入数据库
- 对于已经获取过的postId不要再请求vivo摄影服务器，避免造成负担
- 对于没有创建database的创建database，没有table的创建对应的table
import os
from dotenv import load_dotenv
import requests
from sql import MySQLDatabase
import time
import logging
# 加载 .env 文件
load_dotenv()
class VivoGalleryDB:
def __init__(self):
self.userId = os.getenv('VIVO_USER_ID')
self.host = os.getenv('DB_HOST')
self.port = int(os.getenv('DB_PORT'))
self.database = os.getenv('DB_NAME')
self.user = os.getenv('DB_USER')
self.password = os.getenv('DB_PASSWORD')
def db_connect(self):
try:
self.db = MySQLDatabase(host=self.host, port=self.port, database=self.database, user=self.user, password=self.password)
self.db.connect()
self.conn = self.db.get_connection()
self.cursor = self.conn.cursor()
self.create_tables_if_not_exists()
except Exception as e:
logging.error(f"❌ 无法连接到数据库: {e}")
return None
def create_tables_if_not_exists(self):
try:
self.cursor.execute("""
CREATE TABLE IF NOT EXISTS posts (
post_id VARCHAR(20) PRIMARY KEY,
title VARCHAR(255),
description TEXT,
user_nick VARCHAR(45),
signature VARCHAR(255),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
self.cursor.execute("""
CREATE TABLE IF NOT EXISTS images (
image_id BIGINT(20) AUTO_INCREMENT PRIMARY KEY,
post_id VARCHAR(20),
url VARCHAR(2083),
FOREIGN KEY (post_id) REFERENCES posts(post_id),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
self.conn.commit()
logging.info("✅ 表已创建或已存在。")
except Exception as err:
logging.error(f"❌ 创建表失败: {err}")
def fetch_posts(self):
pageNo = 1
all_data = []
while True:
data = self.fetch_data(pageNo)
if not data:
break
all_data.extend(data)
pageNo += 1
if all_data:
post_ids = [item['postId'] for item in all_data] # 提取每个元素的 postId
logging.info(f'✅找到 {len(post_ids)} 个相册。')
return post_ids
else:
logging.info("❌ 没有发现相册.")
return
def fetch_data(self,pageNo):
base_url = f'https://gallery.vivo.com.cn/gallery/wap/share/user/post/list/{self.userId}.do'
current_time = int(time.time() * 1000) # 获取当前时间的时间戳（毫秒级）
params = {
"dataFrom": 1,
"pageNo": pageNo,
"requestTime": current_time,
"searchType": 4,
"t": current_time
}
headers = {}
try:
with requests.Session() as session:
response = session.get(base_url, headers=headers, params=params)
response.raise_for_status() # 检查请求是否成功
rq = response.json()
if "data" in rq:
data = rq["data"]['posts']
return data # 假设返回的数据是 JSON 格式
else:
return None
except requests.exceptions.RequestException as e:
logging.error(f"❌ 请求失败: {e}")
return None
def save_albums(self,post_ids):
# 批量查询是否存在
placeholders = ', '.join(['%s'] * len(post_ids))
query = f"SELECT post_id FROM posts WHERE post_id IN ({placeholders})"
self.cursor.execute(query, post_ids)
existing_post_ids = [row[0] for row in self.cursor.fetchall()]
url = "https://gallery.vivo.com.cn/gallery/wap/H5/post/getPostDetailById.do"
headers = {
'Content-Type': 'application/x-www-form-urlencoded'
}
for post_id in post_ids:
if str(post_id) not in existing_post_ids:
params = {
"postId": post_id
}
try:
with requests.Session() as session:
response = session.post(url, headers=headers, params=params)
response.raise_for_status() # 检查请求是否成功
rq = response.json()
try:
post = rq["data"]['post']
post_id = post['postId']
title = post.get('postTitle',None)
description = post.get('postDesc',None)
user_nick = post.get('userNick',None)
signature = post.get('signature',None)
urls = post.get('images',[])
#数据库操作
self.cursor.execute("INSERT INTO posts (post_id, title, description,user_nick,signature) VALUES (%s, %s, %s,%s,%s)", (post_id, title, description,user_nick,signature))
for image_url in urls:
self.cursor.execute("INSERT INTO images (post_id, url) VALUES (%s, %s)", (post_id, image_url))
logging.info(f"✅ 新增相册: {post_id} ✅ 相册: {title} ✅ 相册描述: {description}🎉 ✅ 照片: {len(urls)}张🎉")
self.conn.commit()
except Exception as e:
logging.error(f"❌ 处理数据失败: {e}")
except requests.exceptions.RequestException as e:
logging.error(f"❌ 请求失败: {e}")
else:
logging.info(f"❗️ 相册 {post_id} 已存在，跳过。")
def main():
vivo = VivoGalleryDB()
vivo.db_connect()
post_ids = vivo.fetch_posts()
vivo.save_albums(post_ids)
logging.info(f"📢 影相册已保存到数据库。")
vivo.cursor.close()
vivo.conn.close()
vivo.db.disconnect()
logging.info(f"📢 数据库已关闭。")
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
main()
自动化任务可以使用GitHub Actions或青龙面板来定期执行，以确保您的个人作品集始终是最新的。
下一次让我们一起继续这段从快门到网页的奇妙旅程吧！