toolkit-frame之toolkit-sprider 防止重复采集的数据库(Python)

本文涉及的产品
云数据库 RDS MySQL Serverless,0.5-2RCU 50GB
简介: toolkit-frame之toolkit-sprider 防止重复采集的数据库(Python)

对于采集的图片等资源如果采集过了再一次采集就不对了,浪费硬盘也浪费采集的时间,各种浪费,针对这个问题做了个数据库设计,防止同样的数据二次采集。

因为是pytho语言所以框架采用SQLAlchemy。这玩意网上介绍有很多,洒家就不介绍了。不清楚的问问度娘。看代码吧。

from sqlalchemy import Column
from sqlalchemy.types import  String
from main.base.BaseEntity import BaseEntity
class SpriderEntity(BaseEntity):
    __tablename__ = 'sprider_pic'
    sprider_base_url = Column(String, primary_key=True)
    sprider_url = Column(String)
    sprider_pic_index = Column(String)
    sprider_pic_title = Column(String)
    sprider_pager_index = Column(String)
    sprider_type = Column(String)
    create_datetime = Column(String)

当时目光短浅了以为只能采集pic就把名写成PIC了。

另外一个核心就是数据的保存和查询,不废话直接代码:

from sqlalchemy import func
from main.base.BaseFrame import BaseFrame
from main.base.SqliteAccess import SqliteAccess
from main.base.MySqlAccess import MySqlAccess
import pymysql
from sprider.object.SpriderEntity import SpriderEntity
from main.plugin.log.Logger import Logger
pymysql.install_as_MySQLdb()
class SpriderAccess(SqliteAccess):
    # region 保存采集信息 zhangyu-2019-7-18
    def save_sprider(self, model_entity):
        session = SqliteAccess.connection()
        try:
            session.add(model_entity)
            session.commit()
            return True
        except Exception as e:
            Logger.error("save_sprider:保存数据出现 错误" + str(e))
            session.rollback()
            return False
        finally:
            session.close()
    # endregion
    def query_sprider_entity_by_urlandindex(self, sprider_url, pic_index):
        session = SqliteAccess.connection()
        try:
            entity = session.query(SpriderEntity).filter_by(sprider_url=str(sprider_url),
                                                            sprider_pic_index=str(pic_index)).first()
            return entity
        except Exception as e:
            BaseFrame.__err__("query_sprider" + str(e))
            return None
        finally:
            session.close()
    def query_sprider_entity_by_urlandtitle(self, sprider_url, sprider_title):
        session = SqliteAccess.connection()
        try:
            entity = session.query(SpriderEntity).filter_by(sprider_url=sprider_url,
                                                            sprider_pic_title=sprider_title).first()
            return entity
        except Exception as e:
            BaseFrame.__err__("query_sprider_entity_by_urlandtitle"+str(e))
            return None
        finally:
            session.close()
    def delete_sprider_info(self, sprider_url, sprider_title):
        session = MySqlAccess.connection()
        try:
            lottery_entity = session.query(SpriderEntity).filter_by(sprider_url=sprider_url,
                                                                    sprider_pic_title=sprider_title).first()
            if lottery_entity is None:
                return True
            session.delete(lottery_entity)
            session.commit()
            return True
        except:
            session.rollback()
            return False
        finally:
            session.close()
        pass
    def find_pager_by_url_and_type(self, sprider_base_url, sprider_type):
        session = SqliteAccess.connection()
        try:
            # entity = session.query(SpriderEntity).filter_by(sprider_base_url=sprider_base_url,
            #                                                 sprider_type=sprider_type).max()
            entity = session.query(func.max(SpriderEntity.sprider_pager_index)).filter_by(
                sprider_base_url=sprider_base_url, sprider_type=sprider_type).group_by(
                SpriderEntity.sprider_pager_index).all()
            if len(entity) == 0:
                number = 0
            else:
                for e in entity:
                    number = (e[0])
            return int(number)
        except Exception as e:
            Logger.error(str(e))
            return 0
        finally:
            session.close()
        pass
    def query_sprider(self):
        session = SqliteAccess.connection()
        try:
            sprider_list = session.execute("SELECT * FROM sprider_pic LIMIT 10 OFFSET 1").fetchall()
            return sprider_list
        except Exception as e:
            return None, str(e)
        finally:
            session.close()
        pass

里面有一些我框架的东西还望见谅!

 

数据库是sqlit3的,原始文件我就不上传了,自己动手搞搞,锻炼一下。

相关实践学习
基于CentOS快速搭建LAMP环境
本教程介绍如何搭建LAMP环境,其中LAMP分别代表Linux、Apache、MySQL和PHP。
全面了解阿里云能为你做什么
阿里云在全球各地部署高效节能的绿色数据中心,利用清洁计算为万物互联的新世界提供源源不断的能源动力,目前开服的区域包括中国(华北、华东、华南、香港)、新加坡、美国(美东、美西)、欧洲、中东、澳大利亚、日本。目前阿里云的产品涵盖弹性计算、数据库、存储与CDN、分析与搜索、云通信、网络、管理与监控、应用服务、互联网中间件、移动服务、视频服务等。通过本课程,来了解阿里云能够为你的业务带来哪些帮助     相关的阿里云产品:云服务器ECS 云服务器 ECS(Elastic Compute Service)是一种弹性可伸缩的计算服务,助您降低 IT 成本,提升运维效率,使您更专注于核心业务创新。产品详情: https://www.aliyun.com/product/ecs
目录
相关文章
|
3天前
|
NoSQL MongoDB Redis
Python与NoSQL数据库(MongoDB、Redis等)面试问答
【4月更文挑战第16天】本文探讨了Python与NoSQL数据库(如MongoDB、Redis)在面试中的常见问题,包括连接与操作数据库、错误处理、高级特性和缓存策略。重点介绍了使用`pymongo`和`redis`库进行CRUD操作、异常捕获以及数据一致性管理。通过理解这些问题、易错点及避免策略,并结合代码示例,开发者能在面试中展现其技术实力和实践经验。
23 8
Python与NoSQL数据库(MongoDB、Redis等)面试问答
|
3天前
|
SQL 关系型数据库 MySQL
Python与MySQL数据库交互:面试实战
【4月更文挑战第16天】本文介绍了Python与MySQL交互的面试重点,包括使用`mysql-connector-python`或`pymysql`连接数据库、执行SQL查询、异常处理、防止SQL注入、事务管理和ORM框架。易错点包括忘记关闭连接、忽视异常处理、硬编码SQL、忽略事务及过度依赖低效查询。通过理解这些问题和提供策略,可提升面试表现。
22 6
|
9天前
|
SQL 关系型数据库 数据库
Python中SQLite数据库操作详解:利用sqlite3模块
【4月更文挑战第13天】在Python编程中,SQLite数据库是一个轻量级的关系型数据库管理系统,它包含在一个单一的文件内,不需要一个单独的服务器进程或操作系统级别的配置。由于其简单易用和高效性,SQLite经常作为应用程序的本地数据库解决方案。Python的内置sqlite3模块提供了与SQLite数据库交互的接口,使得在Python中操作SQLite数据库变得非常容易。
|
14天前
|
关系型数据库 MySQL 数据库连接
Python+SQLite数据库实现服务端高并发写入
Python中使用SQLite内存模式实现高并发写入:创建内存数据库连接,建立表格,通过多线程并发写入数据。虽然能避免数据竞争,但由于SQLite内存模式采用锁机制,可能在高并发时引发性能瓶颈。若需更高性能,可选择MySQL或PostgreSQL。
18 0
|
20天前
|
JSON 网络协议 前端开发
【UR六轴机械臂源码】python脱离示教器控制UR机械臂实时采集机器人位姿(优傲机器人)
【UR六轴机械臂源码】python脱离示教器控制UR机械臂实时采集机器人位姿(优傲机器人)
|
24天前
|
缓存 NoSQL 关系型数据库
在Python Web开发过程中:数据库与缓存,MySQL和NoSQL数据库的主要差异是什么?
MySQL是关系型DB,依赖预定义的表格结构,适合结构化数据和复杂查询,但扩展性有限。NoSQL提供灵活的非结构化数据存储(如JSON),无统一查询语言,但能横向扩展,适用于大规模、高并发场景。选择取决于应用需求和扩展策略。
113 1
|
1月前
|
数据采集 JSON API
如何实现高效率超简洁的实时数据采集?——Python实战电商数据采集API接口
你是否曾为获取重要数据而感到困扰?是否因为数据封锁而无法获取所需信息?是否因为数据格式混乱而头疼?现在,所有这些问题都可以迎刃而解。让我为大家介绍一款强大的数据采集API接口。
|
1月前
|
SQL Java 关系型数据库
在Python中编写Java数据库驱动是不可能的
在Python中编写Java数据库驱动是不可能的
|
1月前
|
XML 关系型数据库 MySQL
python将word(doc或docx)的内容导入mysql数据库
用python先把doc文件转换成docx文件(这一步也可以不要后续会说明),然后读取docx的文件并另存为htm格式的文件(上一步可以直接把doc文件另存为htm),python根据bs4获取p标签里的内容,如果段落中有图片则保存图片。(图片在word文档中的位置可以很好的还原到生成的数据库内容) 我见网上有把docx压缩后解压获取图片的,然后根据在根据xml来读取图片的位置,我觉得比较繁琐。用docx模块读取段落的时候还需要是不是判断段落中有分页等,然而转成htm之后就不用判断那么多直接判断段落里的样式或者图片等就可以了。
22 1
|
1月前
|
SQL 关系型数据库 数据库连接
描述一下在 Python 中使用数据库的常见方法。
【2月更文挑战第26天】【2月更文挑战第91篇】描述一下在 Python 中使用数据库的常见方法。