python爬取照片存入mysql_python爬取微博图片数据存到Mysql中遇到的各种坑\pythonMysql存储图片...

本人长期出售超大量微博数据、旅游网站评论数据&＃xff0c;并提供各种指定数据爬取服务&＃xff0c;Message to YuboonaZhang&＃64;Yahoo.com。同时欢迎加入社交媒体数据交流群&＃xff1a;99918768

前言

由于硬件等各种原因需要把大概170多万2t左右的微博图片数据存到Mysql中.之前存微博数据一直用的非关系型数据库mongodb&＃xff0c;由于对Mysql的各种不熟悉&＃xff0c;踩了无数坑&＃xff0c;来来回回改了3天才完成。

挖坑填坑之旅

建表

存数据的时候首先需要设计数据库,我准备设计了3个表

微博表&＃xff1a;[id, userid, blog_text, lat, lng, created_time, reserve]   pkey: id

图片表&＃xff1a;[md5, pic_url, pic_bin, exif, reserve]   pkey: md5

关系表&＃xff1a;[id, md5, reserve]   pkey: (id, md5)   fkey: (id, 微博表id)  (md5, 图片表md5)

建表的时候别的问题都还好&＃xff0c;主要是 pic_bin 的类型和 blog_text 的类型有很大的问题&＃xff0c;首先是pic_bin的类型&＃xff0c;开始设置的为BLOB&＃xff0c;但是运行之后发现BLOB最大只能存1M的数据&＃xff0c;并不能满足微博图片的存储&＃xff0c;后改成MEDIUMBLOB(16M)基本能够满足要求了。再后来就是blog_text&＃xff0c;我遇到的第一个大坑

开始的时候很自然的设置blog_text的类型为TEXT&＃xff0c;但跑起来发现有些数据存不进去&＃xff0c;会报错&＃xff0c;经筛查发现是有些微博文本中包含了emoji表情...随后找了很多资料发现是因为utf8下文字是三字节&＃xff0c;但是emoji是四字节&＃xff0c;需要将编码改成utf8mb4。然而我在mac上整mysql的配置文件报各种奇葩错误,一怒之下把TEXT改成了BLOB&＃xff0c;就好了。因为本地是MAC&＃xff0c;我要连接到远程的一台Windows上才能通过那个Windows连接到群晖的Mysql上...本地配置改了也白改。

存图片

然后这就是一个大坑&＃xff01;&＃xff01;&＃xff01; 由于我使用的python3&＃xff0c;所以读取图片得到的二进制的结果前面会有一个b&＃39;, 表示bytes&＃xff0c;正是由于这个b&＃39;导致sql语句拼接的时候这个b后面的单引号会和sql语句的引号结合&＃xff0c;导致后面的二进制没有在引号里面出错&＃xff01;二进制编码又不像string可以对字符转义&＃xff0c;试了好多方法都不行&＃xff01;最后没有办法使用base64 对二进制进行加密转化成字符串&＃xff0c;存到数据库中&＃xff0c;然后要用时的时候再解密。

pic_bin &＃61; str(base64.b64encode(pic_bin))[2:-1]

改配置文件

由于使用Python多进程,一个小时8G数据量&＃xff0c;图片数据比较大&＃xff0c;发包的时候回超过mysql的默认限制&＃xff0c;出现Mysql server has gone away, 这个时候要改配置文件&＃xff0c;在配置文件中参数

max_allowed_packet &＃61; 600M

wait_timeout &＃61; 60000

Lost connection to Mysql server during query

程序跑着跑着总会出现这个错误&＃xff0c;一直找原因&＃xff0c;试了各种办法看了好多资料&＃xff0c;一直都是错误。实在不知道什么原因了...后来一想&＃xff0c;我管他什么原因&＃xff0c;失去连接之后重新连接就行了。使用conn.Ping(True) 判断是否连接mysql成功。如果失去连接就重新连接就行了&＃xff01;最后解决了这个问题

代码实现

#!/usr/bin/env python

# -*- coding: utf-8 -*-

# Created by Baoyi on 2017/10/16

from multiprocessing.pool import Pool

import pymysql

import requests

import json

import exifread

from io import BytesIO

import configparser

import hashlib

import logging

import base64

# 配置logging

logging.basicConfig(level&＃61;logging.WARNING,

format&＃61;&＃39;%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s&＃39;,

datefmt&＃61;&＃39;%a, %d %b %Y %H:%M:%S&＃39;,

filename&＃61;&＃39;weibo.log&＃39;,

filemode&＃61;&＃39;w&＃39;)

cf &＃61; configparser.ConfigParser()

cf.read("ConfigParser.conf")

# 读取配置mysql

db_host &＃61; cf.get("mysql", "db_host")

db_port &＃61; cf.getint("mysql", "db_port")

db_user &＃61; cf.get("mysql", "db_user")

db_pass &＃61; cf.get("mysql", "db_pass")

db &＃61; cf.get("mysql", "db")

# 创建连接

conn &＃61; pymysql.connect(host&＃61;db_host, user&＃61;db_user, passwd&＃61;db_pass, db&＃61;db, port&＃61;db_port, charset&＃61;&＃39;utf8&＃39;)

# 获取游标

cursor &＃61; conn.cursor()

# 创建insert_sql

insert_blog_sql &＃61; (

"INSERT IGNORE INTO blog(userid, id, blog_text, lat, lng, created_time) VALUES(&＃39;{uid}&＃39;, &＃39;{id}&＃39;,&＃39;{blog_text}&＃39;,&＃39;{lat}&＃39;,&＃39;{lng}&＃39;,&＃39;{created_time}&＃39;)"

)

insert_pic_sql &＃61; (

"INSERT IGNORE INTO pics(pic_url, pic_bin, md5, exif) VALUES (&＃39;{pic_url}&＃39;,&＃39;{pic_bin}&＃39;,&＃39;{md5}&＃39;,&＃39;{exif}&＃39;)"

)

insert_relationship_sql &＃61; (

"INSERT IGNORE INTO relationship(id, md5) VALUES (&＃39;{id}&＃39;,&＃39;{md5}&＃39;)"

)

uid &＃61; []

with open(&＃39;./data/final_id.txt&＃39;, &＃39;r&＃39;) as f:

for i in f.readlines():

uid.append(i.strip(&＃39;\r\n&＃39;))

# 处理图片数据

def handle_pic(pic_url):

large_pic_url &＃61; pic_url.replace(&＃39;thumbnail&＃39;, &＃39;large&＃39;)

large_bin &＃61; requests.get(large_pic_url)

return large_bin.content

def get_poiid_info(uid):

try:

url &＃61; &＃39;https://api.weibo.com/2/statuses/user_timeline.json&＃39;

load &＃61; {

&＃39;access_token&＃39;: &＃39;xxxxxxxxxx&＃39;,

&＃39;uid&＃39;: uid,

&＃39;count&＃39;: 100,

&＃39;feature&＃39;: 2,

&＃39;trim_user&＃39;: 1

}

get_info &＃61; requests.get(url&＃61;url, params&＃61;load, timeout&＃61;(10, 10))

if get_info.status_code !&＃61; 200:

logging.warning(ConnectionError)

pass

info_json &＃61; json.loads(get_info.content)

info_json[&＃39;uid&＃39;] &＃61; uid

statuses &＃61; info_json[&＃39;statuses&＃39;]

# 处理筛选微博数据

for status in statuses:

id &＃61; status[&＃39;idstr&＃39;]

if status[&＃39;geo&＃39;] is not None:

lat &＃61; status[&＃39;geo&＃39;][&＃39;coordinates&＃39;][0]

lng &＃61; status[&＃39;geo&＃39;][&＃39;coordinates&＃39;][1]

pic_urls &＃61; status[&＃39;pic_urls&＃39;]

# 判断是否在北京

if (115.7

# 若在北京,插入blog数据进库

blog_text &＃61; status[&＃39;text&＃39;].replace(&＃39;\&＃39;&＃39;, &＃39;\&＃39;\&＃39;&＃39;)

created_time &＃61; status[&＃39;created_at&＃39;]

try:

cursor.execute(

insert_blog_sql.format(uid&＃61;uid, id&＃61;id, blog_text&＃61;blog_text, lat&＃61;lat, lng&＃61;lng,

created_time&＃61;created_time))

except pymysql.err.OperationalError as e_blog:

logging.warning(e_blog.args[1])

pass

# conn.commit()

# 处理图片

for pic_url in pic_urls:

# 获取原图片二进制数据

pic_bin &＃61; handle_pic(pic_url[&＃39;thumbnail_pic&＃39;])

# 读取exif 数据

pic_file &＃61; BytesIO(pic_bin) # 将二进制数据转化成文件对象便于读取exif数据信息和生成MD5

tag1 &＃61; exifread.process_file(pic_file, details&＃61;False, strict&＃61;True)

tag &＃61; {}

for key, value in tag1.items():

if key not in (

&＃39;JPEGThumbnail&＃39;, &＃39;TIFFThumbnail&＃39;, &＃39;Filename&＃39;,

&＃39;EXIF MakerNote&＃39;): # 去除四个不必要的exif属性&＃xff0c;简化信息量

tag[key] &＃61; str(value)

tags &＃61; json.dumps(tag) # dumps为json类型 此tag即为exif的json数据

# 生成MD5

MD5 &＃61; hashlib.md5(pic_file.read()).hexdigest()

# 首先把二进制图片用base64 转成字符串之后再存

try:

cursor.execute(

insert_pic_sql.format(pic_url&＃61;pic_url[&＃39;thumbnail_pic&＃39;].replace(&＃39;thumbnail&＃39;, &＃39;large&＃39;),

pic_bin&＃61;str(base64.b64encode(pic_bin))[2:-1], md5&＃61;MD5,

exif&＃61;tags))

except pymysql.err.OperationalError as e_pic:

logging.warning(e_pic.args[1])

pass

try:

cursor.execute(insert_relationship_sql.format(id&＃61;id, md5&＃61;MD5))

except pymysql.err.OperationalError as e_relation:

logging.warning(e_relation)

pass

conn.commit()

else:

logging.info(id &＃43; " is Not in Beijing")

pass

else:

logging.info(id &＃43; &＃39; Geo is null&＃39;)

pass

except pymysql.err.OperationalError as e:

logging.error(e.args[1])

pass

def judge_conn(i):

global conn

try:

conn.ping(True)

get_poiid_info(i)

except pymysql.err.OperationalError as e:

logging.error(&＃39;Reconnect&＃39;)

conn &＃61; pymysql.connect(host&＃61;db_host, user&＃61;db_user, passwd&＃61;db_pass, db&＃61;db, charset&＃61;&＃39;utf8&＃39;)

get_poiid_info(i)

def handle_tuple(a_tuple):

read_uid_set &＃61; []

for i in a_tuple:

read_uid_set.append(i[0])

return set(read_uid_set)

if __name__ &＃61;&＃61; &＃39;__main__&＃39;:

sql_find_uid &＃61; (

"SELECT userid FROM blog"

)

cursor.execute(sql_find_uid)

read_uid_tuple &＃61; cursor.fetchall()

read_list &＃61; handle_tuple(read_uid_tuple)

print(len(read_list))

new_uid &＃61; set(uid).difference(read_list)

print(len(new_uid))

pool &＃61; Pool()

pool.map(judge_conn, list(new_uid))

个人博客