使用sqlite，sqlalchemy，python在數據庫中重複插入

Question

我正在學習Python，並且通過在線資源和此站點上的人員的幫助，我正在掌握它。在我的第一個腳本中，我正在解析Twitter RSS源條目並將結果插入到數據庫中，還有一個我無法修復的問題。即，重複的條目正被插入到其中一個表中。

作爲一個背景，我最初在HalOtis.com上發現了一個用於下載RSS源的基本腳本，然後通過幾種方式對其進行了修改：1）修改了Twitter RSS源中的特性（未將其分離爲內容，標題，網址等）; 2）添加了「hashtags」和多對多關係（entry_tag表）的表格; 3）將表格設置更改爲sqlalchemy; 4）進行了一些臨時更改，以解決正在發生的奇怪unicode問題。因此，代碼在各個地方都很難看，但它已經成爲一種很好的學習體驗，現在可以工作 - 除了它會在「條目」表中插入重複內容。

因爲我不確定什麼對人最有幫助，所以我粘貼了下面的所有代碼，並在一些地方發表了一些評論，指出我認爲最重要的內容。

我真的很感謝任何幫助。謝謝！

編輯：有人建議我爲數據庫提供模式。我從來沒有這樣做過，所以如果我做得不對，請忍受我。我設置了四個表：

1. RSSFeeds，其中包含Twitter的RSS Feed列表
2. RSSEntries，其中包含從每個飼料的下載（分析後）個別條目的列表（與內容列，井號標籤，日期，URL）
3. 標籤，它包含了所有在每個條目（鳴叫）
4. entry_tag，其中包含讓我的標籤映射到項列中發現的井號標籤的列表。

總之，以下腳本從RSS Feeds表中抓取了五個測試RSS源，從每個源下載20個最新的條目/推文，解析條目，並將信息放入RSS條目，標籤，和entry_tag表。

#!/usr/local/bin/python 

import sqlite3 
import threading 
import time 
import Queue 
from time import strftime 
import re  
from string import split 
import feedparser 
from django.utils.encoding import smart_str, smart_unicode  
from sqlalchemy import schema, types, ForeignKey, select, orm 
from sqlalchemy import create_engine 

engine = create_engine('sqlite:///test98.sqlite', echo=True) 
metadata = schema.MetaData(engine) 
metadata.bind = engine 

def now(): 
    return datetime.datetime.now() 


#set up four tables, with many-to-many relationship 
RSSFeeds = schema.Table('feeds', metadata, 
    schema.Column('id', types.Integer, 
     schema.Sequence('feeds_seq_id', optional=True), primary_key=True), 
    schema.Column('url', types.VARCHAR(1000), default=u''), 
) 


RSSEntries = schema.Table('entries', metadata, 
    schema.Column('id', types.Integer, 
     schema.Sequence('entries_seq_id', optional=True), primary_key=True), 
    schema.Column('feed_id', types.Integer, schema.ForeignKey('feeds.id')), 
    schema.Column('short_url', types.VARCHAR(1000), default=u''), 
    schema.Column('content', types.Text(), nullable=False), 
    schema.Column('hashtags', types.Unicode(255)), 
    schema.Column('date', types.String()), 
) 


tag_table = schema.Table('tag', metadata, 
    schema.Column('id', types.Integer, 
     schema.Sequence('tag_seq_id', optional=True), primary_key=True), 
    schema.Column('tagname', types.Unicode(20), nullable=False, unique=True), 
) 


entrytag_table = schema.Table('entrytag', metadata, 
    schema.Column('id', types.Integer, 
     schema.Sequence('entrytag_seq_id', optional=True), primary_key=True), 
    schema.Column('entryid', types.Integer, schema.ForeignKey('entries.id')), 
    schema.Column('tagid', types.Integer, schema.ForeignKey('tag.id')), 
) 


metadata.create_all(bind=engine, checkfirst=True) 


# Insert test set of Twitter RSS feeds 
stmt = RSSFeeds.insert() 
stmt.execute(
    {'url': 'http://twitter.com/statuses/user_timeline/14908909.rss'}, 
    {'url': 'http://twitter.com/statuses/user_timeline/52903246.rss'}, 
    {'url': 'http://twitter.com/statuses/user_timeline/41902319.rss'}, 
    {'url': 'http://twitter.com/statuses/user_timeline/29950404.rss'}, 
    {'url': 'http://twitter.com/statuses/user_timeline/35699859.rss'}, 
) 



#These 3 lines for threading process (see HalOtis.com for example) 
THREAD_LIMIT = 20 
jobs = Queue.Queue(0) 
rss_to_process = Queue.Queue(THREAD_LIMIT) 


#connect to sqlite database and grab the 5 test RSS feeds 
conn = engine.connect() 
feeds = conn.execute('SELECT id, url FROM feeds').fetchall() 

#This block contains all the parsing and DB insertion 
def store_feed_items(id, items): 
    """ Takes a feed_id and a list of items and stores them in the DB """ 
    for entry in items: 
     conn.execute('SELECT id from entries WHERE short_url=?', (entry.link,)) 
     #note: entry.summary contains entire feed entry for Twitter, 
        #i.e., not separated into content, etc. 
     s = unicode(entry.summary) 
     test = s.split() 
     tinyurl2 = [i for i in test if i.startswith('http://')] 
     hashtags2 = [i for i in s.split() if i.startswith('#')] 
     content2 = ' '.join(i for i in s.split() if i not in tinyurl2+hashtags2) 
     content = unicode(content2) 
     tinyurl = unicode(tinyurl2) 
     hashtags = unicode (hashtags2) 
     print hashtags 
     date = strftime("%Y-%m-%d %H:%M:%S",entry.updated_parsed) 


     #Insert parsed feed data into entries table 
        #THIS IS WHERE DUPLICATES OCCUR 
     result = conn.execute(RSSEntries.insert(), {'feed_id': id, 'short_url': tinyurl, 
      'content': content, 'hashtags': hashtags, 'date': date}) 
     entry_id = result.last_inserted_ids()[0] 


     #Look up tag identifiers and create any that don't exist: 
     tags = tag_table 
     tag_id_query = select([tags.c.tagname, tags.c.id], tags.c.tagname.in_(hashtags2)) 
     tag_ids = dict(conn.execute(tag_id_query).fetchall()) 
     for tag in hashtags2: 
      if tag not in tag_ids: 
       result = conn.execute(tags.insert(), {'tagname': tag}) 
       tag_ids[tag] = result.last_inserted_ids()[0] 

     #insert data into entrytag table 
     if hashtags2: conn.execute(entrytag_table.insert(), 
      [{'entryid': entry_id, 'tagid': tag_ids[tag]} for tag in hashtags2]) 


#Rest of file completes the threading process  
def thread(): 
    while True: 
     try: 
      id, feed_url = jobs.get(False) # False = Don't wait 
     except Queue.Empty: 
      return 

     entries = feedparser.parse(feed_url).entries 
     rss_to_process.put((id, entries), True) # This will block if full 

for info in feeds: # Queue them up 
    jobs.put([info['id'], info['url']]) 

for n in xrange(THREAD_LIMIT): 
    t = threading.Thread(target=thread) 
    t.start() 

while threading.activeCount() > 1 or not rss_to_process.empty(): 
    # That condition means we want to do this loop if there are threads 
    # running OR there's stuff to process 
    try: 
     id, entries = rss_to_process.get(False, 1) # Wait for up to a second 
    except Queue.Empty: 
     continue 

    store_feed_items(id, entries) 

Answer 1

看起來你SQLAlchemy的包括成不使用SQLAlchemy以前存在的腳本。這裏有太多的移動部件，我們沒有一個明顯理解得足夠好。

我會建議從頭開始。不要使用線程。不要使用sqlalchemy。要開始也許不要使用SQL數據庫。使用簡單的循環和可能的time.sleep（），將一個腳本收集到您想要的信息中，並以簡單的方式將其收集到一個簡單的數據結構中。然後，當它工作時，您可以將存儲添加到SQL數據庫中，並且我真的不認爲直接編寫SQL語句比使用ORM困難得多，並且更容易調試恕我直言。很有可能你永遠不需要添加線程。

「如果您認爲自己足夠聰明，可以編寫多線程程序，那麼您就不是。」 - James Ahlstrom。