Python pymongo.errors 模块，DuplicateKeyError() 实例源码

def __flush_data(self, data):
        """Flush `data` to a chunk.
        """
        # Ensure the index, even if there's nothing to write, so
        # the filemd5 command always succeeds.
        self._ensure_index()

        if not data:
            return
        assert(len(data) <= self.chunk_size)

        chunk = {"files_id": self._file["_id"],
                 "n": self._chunk_number,
                 "data": Binary(data)}

        try:
            self._chunks.insert(chunk)
        except DuplicateKeyError:
            self._raise_file_exists(self._file['_id'])
        self._chunk_number += 1
        self._position += len(data)

def __flush_data(self, data):
        """Flush `data` to a chunk.
        """
        # Ensure the index, even if there's nothing to write, so
        # the filemd5 command always succeeds.
        self._ensure_index()

        if not data:
            return
        assert(len(data) <= self.chunk_size)

        chunk = {"files_id": self._file["_id"],
                 "n": self._chunk_number,
                 "data": Binary(data)}

        try:
            self._chunks.insert(chunk)
        except DuplicateKeyError:
            self._raise_file_exists(self._file['_id'])
        self._chunk_number += 1
        self._position += len(data)

def __flush_data(self, data):
        """Flush `data` to a chunk.
        """
        # Ensure the index, even if there's nothing to write, so
        # the filemd5 command always succeeds.
        self._ensure_index()

        if not data:
            return
        assert(len(data) <= self.chunk_size)

        chunk = {"files_id": self._file["_id"],
                 "n": self._chunk_number,
                 "data": Binary(data)}

        try:
            self._chunks.insert(chunk)
        except DuplicateKeyError:
            self._raise_file_exists(self._file['_id'])
        self._chunk_number += 1
        self._position += len(data)

def put_url(self, url, title, timestamp=0.0, filetype=""):
        # ????url??
        suffix = url.split(".")[-1].split("?")[0]
        if not filetype:
            filetype = self.conf['list']['api']['filetype']
            for i in self.file_suffix:
                if i == suffix:
                    filetype = "file"
                    break
        # ??????
        if timestamp:
            obj = {"title": title, "timestamp": timestamp, "_id": url, "type": filetype}
        else:
            obj = {"title": title, "_id": url, "type": filetype}

        try:
            return self.con.insert(obj)
        except errors.DuplicateKeyError as dk:

            raise dk
        except Exception as e:
            raise e

    # ??url??

def store_article_list(self, data):
        data = json.loads(data)
        # list.do
        total_page = data['data']['totalPages']
        config['basic']['total_page'] = total_page
        for item in data['data']['content']:
            html_url = "http://www.neeq.com.cn" + item['htmlUrl']
            title = item['title']
            t = item['publishDate']
            timestamp = Util.get_timestamp(t, f="%Y-%m-%d %H:%M:%S.0")
            try:
                self.url_manager.put_url(html_url, title, timestamp, "html")
                Util.COUNT_DUPLICATE = 0
                Util.COUNT_SUCCESS += 1
            except errors.DuplicateKeyError as dk:
                Util.COUNT_DUPLICATE += 1
                if config['list']['max_replicate']:
                    if Util.COUNT_DUPLICATE > config['list']['max_replicate']:
                        Util.COUNT_DUPLICATE = 0
                        raise exception.ExceedMaxDuplicate
        Util.view_bar(Util.COUNT_PROCESSED, total_page)
        Util.COUNT_PROCESSED += 1

def store_article(self, res, url):
        if res:
            res['_id'] = url
            print(url, "\t", res['title'])
            try:
                self.db.put_article(res)
                Util.COUNT_DUPLICATE = 0
                Util.COUNT_SUCCESS += 1
            except errors.DuplicateKeyError as dk:
                Util.COUNT_DUPLICATE += 1
                if config['article']['max_replicate']:
                    if Util.COUNT_DUPLICATE > config['article']['max_replicate']:
                        Util.COUNT_DUPLICATE = 0
                        raise exception.ExceedMaxDuplicate
            self.url_manager.set_url_status(url, 2)
        else:
            self.url_manager.set_url_status(url, -1)

def __flush_data(self, data):
        """Flush `data` to a chunk.
        """
        # Ensure the index, even if there's nothing to write, so
        # the filemd5 command always succeeds.
        await self.__ensure_indexes()
        self._file['md5'].update(data)

        if not data:
            return
        assert(len(data) <= self.chunk_size)

        chunk = {'files_id': self._file['_id'],
                 'n': self._chunk_number,
                 'data': Binary(data)}

        try:
            await self._chunks.insert_one(chunk)
        except DuplicateKeyError:
            self._raise_file_exists(self._file['_id'])
        self._chunk_number += 1
        self._position += len(data)

def inc(op: str, ident: str, period_secs: int, max_operations: int):
  coll = db.coll('opcount')
  cur_time = int(time.time())
  begin_at = datetime.datetime.utcfromtimestamp(cur_time - cur_time % period_secs)
  expire_at = begin_at + datetime.timedelta(seconds=period_secs)
  try:
    doc = await coll.find_one_and_update(filter={'ident': ident,
                                                 'begin_at': begin_at,
                                                 'expire_at': expire_at,
                                                 op: {'$not': {'$gte': max_operations}}},
                                         update={'$inc': {op: 1}},
                                         upsert=True,
                                         return_document=ReturnDocument.AFTER)
    return doc
  except errors.DuplicateKeyError:
    raise error.OpcountExceededError(op, period_secs, max_operations)

def add(domain_id: str, owner_uid: int,
              roles=builtin.DOMAIN_SYSTEM['roles'],
              name: str=None, gravatar: str=None, bulletin: str=''):
  validator.check_domain_id(domain_id)
  validator.check_name(name)
  validator.check_bulletin(bulletin)
  for domain in builtin.DOMAINS:
    if domain['_id'] == domain_id:
      raise error.DomainAlreadyExistError(domain_id)
  coll = db.coll('domain')
  try:
    result = await coll.insert_one({'_id': domain_id,
                                    'pending': True, 'owner_uid': owner_uid,
                                    'roles': roles, 'name': name,
                                    'gravatar': gravatar, 'bulletin': bulletin})
    domain_id = result.inserted_id
  except errors.DuplicateKeyError:
    raise error.DomainAlreadyExistError(domain_id) from None
  # grant root role to owner by default
  await add_user_role(domain_id, owner_uid, builtin.ROLE_ROOT)
  await coll.update_one({'_id': domain_id},
                        {'$unset': {'pending': ''}})
  return domain_id

def test_update_safe():
    s = get_session()
    s.clear_collection(TUnique)
    s.query(TUnique).filter_by(i=1).set(i=1, j=2).upsert().execute()
    # default safe=false -- ignore error
    s.query(TUnique).filter_by(i=1).set(i=1, j=2).upsert().execute()
    # explicit safe=false
    s.query(TUnique).filter_by(i=1).set(i=1, j=2).safe().safe(safe=False).upsert().execute()

    # safe=true, exception
    # TODO: doesn't produce a real exception.  should investigate why, but I checked
    # and I am sending safe=True
    #
    # try:
    #     s.query(TUnique).filter_by(i=1).set(i=1, j=2).safe().upsert().execute()
    #     assert False, 'No error raised on safe insert for second unique item'
    # except DuplicateKeyError:
    #     pass


# Test Remove

def __flush_data(self, data):
        """Flush `data` to a chunk.
        """
        # Ensure the index, even if there's nothing to write, so
        # the filemd5 command always succeeds.
        self._ensure_index()

        if not data:
            return
        assert(len(data) <= self.chunk_size)

        chunk = {"files_id": self._file["_id"],
                 "n": self._chunk_number,
                 "data": Binary(data)}

        try:
            self._chunks.insert(chunk)
        except DuplicateKeyError:
            self._raise_file_exists(self._file['_id'])
        self._chunk_number += 1
        self._position += len(data)

def save(self, **kwargs):
        if self.__class__.exist_count > 10:
            raise RuntimeError('Exceed maximum retry times for database')

        try:
            # add source and chinese source
            kwargs['src'] = self.__class__.name
            kwargs['ch_src'] = self.__class__.ch_name

            # make cdn path
            img_url = kwargs['img'].encode("utf-8")
            cdn_path = md5.new(img_url).hexdigest()
            kwargs['cdn_path'] = cdn_path.decode("utf-8")

            # insert into database
            timeline = db.get_collection('timeline')
            timeline.insert(kwargs)

            # actually upload image here
            if not self.__class__.img_store.exists(cdn_path):
                self.__class__.img_store.upload_remote_image(cdn_path, img_url)

        except DuplicateKeyError:
            self.__class__.exist_count += 1

def inc(op: str, id: str, period_secs: int, max_operations: int):
    coll = db.Collection('op_count')
    cur_time = int(time.time())
    begin_at = datetime.datetime.utcfromtimestamp(cur_time - cur_time % period_secs)
    expire_at = begin_at + datetime.timedelta(seconds=period_secs)
    try:
        doc = await coll.find_one_and_update(filter={'id': id,
                                                     'begin_at': begin_at,
                                                     'expire_at': expire_at,
                                                     op: {'$not': {'$gte': max_operations}}},
                                             update={'$inc': {op: 1}},
                                             upsert=True,
                                             return_document=True)
        return doc
    except errors.DuplicateKeyError:
        raise error.OpCountExceededError(op, period_secs, max_operations)

def add(domain_id: str, owner_uid: int,
              roles=builtin.DOMAIN_SYSTEM['roles'],
              name: str=None, gravatar: str=None):
    validator.check_domain_id(domain_id)
    validator.check_name(name)
    for domain in builtin.DOMAINS:
        if domain['_id'] == domain_id:
            raise error.DomainAlreadyExistError(domain_id)
    coll = db.Collection('domain')
    try:
        return await coll.insert({
            '_id': domain_id,
            'owner_uid': owner_uid,
            'roles': roles,
            'name': name,
            'gravatar': gravatar
        })
    except errors.DuplicateKeyError:
        raise error.DomainAlreadyExistError(domain_id) from None

def add(campaign_id: str, title: str, content: str, owner_uid: int,
              begin_at: lambda i: datetime.datetime.utcfromtimestamp(int(i)),
              end_at: lambda i: datetime.datetime.utcfromtimestamp(int(i)),
              is_newbie: bool):
    validator.check_title(title)
    validator.check_content(content)
    validator.check_domain_id(campaign_id)
    if begin_at >= end_at:
        raise error.ValidationError('begin_at', 'end_at')
    coll = db.Collection('campaign')
    try:
        return (await coll.insert_one({'_id': campaign_id,
                                       'domain_id': builtin.DOMAIN_ID_SYSTEM,
                                       'owner_uid': owner_uid,
                                       'title': title,
                                       'content': content,
                                       'is_newbie': is_newbie,
                                       'begin_at': begin_at,
                                       'end_at': end_at})).inserted_id
    except errors.DuplicateKeyError:
        raise error.CampaignAlreadyExistError(campaign_id) from None

def attend(domain_id: str, tid: int, uid: int):
    #  TODO: check time.
    coll = db.Collection('contest.status')
    try:
        await coll.find_one_and_update(filter={'domain_id': domain_id,
                                               'tid': tid,
                                               'uid': uid,
                                               'attend': {'$eq': 0}},
                                       update={'$set': {'attend': 1}},
                                       upsert=True,
                                       return_document=ReturnDocument.AFTER)
    except errors.DuplicateKeyError:
        raise error.ContestAlreadyAttendedError(domain_id, tid, uid) from None
    coll = db.Collection('contest')
    return await coll.find_one_and_update(filter={'domain_id': domain_id,
                                                  '_id': tid},
                                          update={'$inc': {'attend': 1}},
                                          return_document=ReturnDocument.AFTER)

def download_arch_security():
    db = get_db()
    collection = db.arch_security_updates
    collection.create_index([('package', ASCENDING),
                             ('announced_at', ASCENDING)], unique=True)

    for package, dt, source in rss_feed():
        try:
            collection.insert_one({'package': package,
                                   'announced_at': dt,
                                   'source': source})
        except DuplicateKeyError:
            return
        else:
            log.info('Identified Arch security update for {}, '
                     'announced at {}', package, dt)
            yield (package, dt)

def _create_archive(
            self,
            archive_name,
            metadata):

        try:
            self.collection.insert_one(metadata)
        except DuplicateKeyError:
            raise KeyError('Archive "{}" already exists'.format(archive_name))

def store_revision(coll, rev, metadata):
    if '*' not in rev:
        # this revision was deleted
        return None
    d = json.loads(rev['*'])
    del rev['*']
    d.update(rev)
    d['_id'] = d['revid']
    d['metadata'] = metadata if metadata else dict()
    if isinstance(d['timestamp'], time.struct_time):
        d['timestamp'] = datetime.datetime.fromtimestamp(mktime(d['timestamp']))
    elif not isinstance(d['timestamp'], str):
        d['timestamp'] = time.strftime(d['timestamp'], '%Y-%m-%dT%H:%M:%SZ')
    try:
        coll.insert_one(d)
    except DuplicateKeyError:
        pass

def __flush(self):
        """Flush the file to the database.
        """
        try:
            self.__flush_buffer()

            db = self._coll.database

            # See PYTHON-417, "Sharded GridFS fails with exception: chunks out
            # of order." Inserts via mongos, even if they use a single
            # connection, can succeed out-of-order due to the writebackListener.
            # We mustn't call "filemd5" until all inserts are complete, which
            # we ensure by calling getLastError (and ignoring the result).
            db.error()

            md5 = db.command(
                "filemd5", self._id, root=self._coll.name)["md5"]

            self._file["md5"] = md5
            self._file["length"] = self._position
            self._file["uploadDate"] = datetime.datetime.utcnow()

            return self._coll.files.insert(self._file,
                                           **self._coll._get_wc_override())
        except DuplicateKeyError:
            self._raise_file_exists(self._id)

def __flush(self):
        """Flush the file to the database.
        """
        try:
            self.__flush_buffer()

            db = self._coll.database

            # See PYTHON-417, "Sharded GridFS fails with exception: chunks out
            # of order." Inserts via mongos, even if they use a single
            # connection, can succeed out-of-order due to the writebackListener.
            # We mustn't call "filemd5" until all inserts are complete, which
            # we ensure by calling getLastError (and ignoring the result).
            db.error()

            md5 = db.command(
                "filemd5", self._id, root=self._coll.name)["md5"]

            self._file["md5"] = md5
            self._file["length"] = self._position
            self._file["uploadDate"] = datetime.datetime.utcnow()

            return self._coll.files.insert(self._file,
                                           **self._coll._get_wc_override())
        except DuplicateKeyError:
            self._raise_file_exists(self._id)

def handle_save(self, document, callback):
        def handle(*arguments, **kw):
            if len(arguments) > 1 and arguments[1]:
                if isinstance(arguments[1], (DuplicateKeyError, )):
                    raise UniqueKeyViolationError.from_pymongo(str(arguments[1]), self.__klass__)
                else:
                    raise arguments[1]

            document._id = arguments[0]
            callback(document)

        return handle

def fetch_article(self, url):
        try:
            if "html" in config['article']:
                html = self.fetch_html(url)
                Filter(html, config['article']['html']).store("article", url)
            elif "api" in config['article']:
                # TODO ?api????
                pass
            else:
                raise Exception("??????????")
        except errors.DuplicateKeyError as e:
            self.duplicate_count += 1
            print("????", url)

    # ??????

def store_file_list(self, data):
        data = json.loads(data)
        # infoResult.do
        total_page = data['listInfo']['totalPages']
        config['basic']['total_page'] = total_page
        for item in data['listInfo']['content']:
            file_url = "http://www.neeq.com.cn" + item['destFilePath']
            title = item['disclosureTitle']
            timestamp = item['upDate']['time']
            # ??????(??)
            # if config['basic']['max_replicate']:
            #     if timestamp <= self.url_manager.get_last_url("file")['timestamp']:
            #         raise exception.ListFinishedException
            try:
                self.url_manager.put_url(file_url, title, timestamp, "file")
                # ???????????????0
                Util.COUNT_DUPLICATE = 0
                Util.COUNT_SUCCESS += 1
            except errors.DuplicateKeyError as dk:
                # ??????+1
                Util.COUNT_DUPLICATE += 1
                if config['list']['max_replicate']:
                    if Util.COUNT_DUPLICATE > config['list']['max_replicate']:
                        Util.COUNT_DUPLICATE = 0
                        raise exception.ExceedMaxDuplicate
                        # ?????????????????????
        Util.view_bar(Util.COUNT_PROCESSED, total_page)
        Util.COUNT_PROCESSED += 1

def add_tags(self, tags: List[str]):
        for t in tags:
            try:
                self.tags.insert_one({
                    'name': t
                })
            except DuplicateKeyError:
                pass

def __flush(self):
        """Flush the file to the database.
        """
        try:
            await self.__flush_buffer()

            self._file['md5'] = self._file['md5'].hexdigest()
            self._file['length'] = self._position
            self._file['uploadDate'] = datetime.datetime.utcnow()

            return await self._coll.files.insert_one(self._file)
        except DuplicateKeyError:
            self._raise_file_exists(self._id)

def save_stock_list(self, stocks):
        if len(stocks) > 0:
            try:
                stocklist_collection = self.db[self.STOCK_LIST]
                stocklist_collection.insert_many(stocks, ordered=False)
            except (DuplicateKeyError, BulkWriteError) as err:
                logger.error("save_stock_list: %i %s", len(stocks), err)

def save_stock_historical_data(self, quote, stock_historical_data_array):
        if len(stock_historical_data_array) > 0:
            try:
                collection_name = quote + self.HISTORICAL_DATA_SUFIX
                self.create_historical_collection_if_not_exists(collection_name)
                stock_historical_data_collection = self.db[collection_name]
                stock_historical_data_collection.insert_many(stock_historical_data_array, ordered=False)
            except (DuplicateKeyError, BulkWriteError) as err:
                logger.error("save_stock_historical_data: %s %i %s", quote, len(stock_historical_data_array), err)

def upsert_stock_current_data(self, quote, stock_current_data):
        if stock_current_data is not None:
            try:
                stock_current_data_collection = self.db[self.STOCKS_CURRENT_DATA]
                query = { self.SYMBOL_KEY: quote }
                stock_current_data_collection.replace_one(query, stock_current_data, upsert=True)
            except DuplicateKeyError as err:
                logger.error("upsert_stock_current_data: %s %s", quote, err)

def acquire_lock(lock_name: str):
  lock_value = random.randint(1, 0xFFFFFFFF)
  coll = db.coll('system')
  try:
    await coll.update_one(filter={'_id': 'lock_' + lock_name, 'value': 0},
                          update={'$set': {'value': lock_value}},
                          upsert=True)
  except errors.DuplicateKeyError:
    return None
  return lock_value

def enroll(domain_id: str, tid: objectid.ObjectId, uid: int):
  try:
    await document.capped_inc_status(domain_id, document.TYPE_TRAINING, tid,
                                     uid, 'enroll', 1, 0, 1)
  except errors.DuplicateKeyError:
    raise error.TrainingAlreadyEnrollError(domain_id, tid, uid) from None
  return await document.inc(domain_id, document.TYPE_TRAINING, tid, 'enroll', 1)

def vote_solution(domain_id: str, psid: document.convert_doc_id, uid: int, value: int):
  try:
    pssdoc = await document.capped_inc_status(domain_id, document.TYPE_PROBLEM_SOLUTION, psid,
                                              uid, 'vote', value)
  except errors.DuplicateKeyError:
    raise error.AlreadyVotedError(domain_id, psid, uid) from None
  psdoc = await document.inc(domain_id, document.TYPE_PROBLEM_SOLUTION, psid, 'vote', value)
  await domain.inc_user(domain_id, psdoc['owner_uid'], num_liked=value)
  return psdoc, pssdoc

def update_status(domain_id: str, pid: document.convert_doc_id, uid: int,
                        rid: objectid.ObjectId, status: int):
  try:
    return await document.set_if_not_status(domain_id, document.TYPE_PROBLEM, pid, uid,
                                            'status', status, constant.record.STATUS_ACCEPTED,
                                            rid=rid)
  except errors.DuplicateKeyError:
    return None

def _update_nodes(domain_id, nodes):
  items = list(nodes.items())
  try:
    await document.add(domain_id, items, 0, document.TYPE_DISCUSSION_NODE,
                       document.DOC_ID_DISCUSSION_NODES)
  except errors.DuplicateKeyError:
    doc = await document.set(domain_id, document.TYPE_DISCUSSION_NODE,
                             document.DOC_ID_DISCUSSION_NODES, content=items)
    if not doc:
      raise error.InvalidStateError()
  await smallcache.unset_global(smallcache.PREFIX_DISCUSSION_NODES + domain_id)

def add_user_role(domain_id: str, uid: int, role: str, join_at=None):
  validator.check_role(role)
  if join_at is None:
    join_at = datetime.datetime.utcnow()
  coll = db.coll('domain.user')
  try:
    await coll.update_one({'domain_id': domain_id, 'uid': uid,
                           'role': {'$exists': False}},
                           {'$set': {'role': role, 'join_at': join_at}},
                           upsert=True)
  except errors.DuplicateKeyError:
    raise error.UserAlreadyDomainMemberError(domain_id, uid) from None
  return True

def inc_user_usage(domain_id: str, uid: int, usage_field: str, usage: int, quota: int):
  coll = db.coll('domain.user')
  try:
    return await coll.find_one_and_update(filter={'domain_id': domain_id, 'uid': uid,
                                                  usage_field: {'$not': {'$gte': quota - usage}}},
                                          update={'$inc': {usage_field: usage}},
                                          upsert=True,
                                          return_document=ReturnDocument.AFTER)
  except errors.DuplicateKeyError:
    raise error.UsageExceededError(domain_id, uid, usage_field, usage, quota)

def add(uid: int, uname: str, password: str, mail: str, regip: str=''):
  """Add a user."""
  validator.check_uname(uname)
  # TODO(iceboy): Filter uname by keywords.
  validator.check_password(password)
  validator.check_mail(mail)

  uname_lower = uname.strip().lower()
  mail_lower = mail.strip().lower()

  for user in builtin.USERS:
    if user['_id'] == uid or user['uname_lower'] == uname_lower or user['mail_lower'] == mail_lower:
      raise error.UserAlreadyExistError(uname)

  salt = pwhash.gen_salt()
  coll = db.coll('user')
  try:
    await coll.insert_one({'_id': uid,
                           'uname': uname,
                           'uname_lower': uname_lower,
                           'mail': mail,
                           'mail_lower': mail_lower,
                           'salt': salt,
                           'hash': pwhash.hash_vj4(password, salt),
                           'regat': datetime.datetime.utcnow(),
                           'regip': regip,
                           'priv': builtin.DEFAULT_PRIV,
                           'loginat': datetime.datetime.utcnow(),
                           'loginip': regip,
                           'gravatar': mail})
  except errors.DuplicateKeyError:
    raise error.UserAlreadyExistError(uid, uname, mail) from None

def push(self, url):
        """Add new URL to queue if does not exist
        """
        try:
            self.db.crawl_queue.insert({'_id': url, 'status': self.OUTSTANDING})
        except errors.DuplicateKeyError as e:
            pass # this is already in the queue

def push(self, url, title):
        try:
            self.db.insert({'_id': url, 'status': self.COMPLETE, '??': title})
            print(u'??????')
        except errors.DuplicateKeyError as e:
            print('??????:{0}, {1}???????????'.format(e, title))
            pass

def push_imgurl(self, title, url):
        try:
            self.db.insert(
                {'_id': title, 'status': self.OUTSTANDING, 'url': url})
            print(u'????????')
        except errors.DuplicateKeyError as e:
            print('??????:{0}, {1}???????????'.format(e, url))
            pass

def test_safe_with_error():
    s = Session.connect('unit-testing')
    s.clear_collection(TUnique)
    s.save(TUnique(i=1))
    try:
        s.save(TUnique(i=1), safe=True)
        assert False, 'No error raised on safe insert for second unique item'
    except DuplicateKeyError:
        assert len(s.queue) == 0

def storeContent(self, content):
        content['created_time'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        try:
            post_id = self.posts.insert(content)
            return post_id
        except errors.DuplicateKeyError, e:
            raise Exception('err', 'DuplicateKeyError')

def on_data(self, status):
        try:
            json_data = json.loads(status)
            user = json_data["user"]["screen_name"]
            try:
                screen_names.insert_one({"_id": user, "collected": False, "completed": False})

            except DuplicateKeyError:
                logging.info("User {} already exists.".format(user))
                pass

        except KeyError:
            sleep(1000)
            logging.error("Users not stored.")

def push_theme(self,url,title,number):#??????????URL??URL????????
        try:
            self.db.insert({'_id':url,'status':self.OUTSIANDING,'??':title,'????':number})
            print(title,url,'??????')
        except errors.DuplicateKeyError as e:#??????????????
            print(title,url,'???????')
            pass

def push_queue(self,url):
        try:
            self.db.insert({'_id':url,'status':self.OUTSIANDING})
            print(url,'??????')
        except errors.DuplicateKeyError as e:#??????????????
            print(url,'???????')
            pass

def push_book(self,title,author,book_style,book_introduction,book_url):
        try:
            self.db.insert({'_id':book_url,'????':title,'????':author,'????':book_style,'??':book_introduction})
            print(title, '????????')
        except errors.DuplicateKeyError as e:
            print(title, '?????????')
            pass

def __flush(self):
        """Flush the file to the database.
        """
        try:
            self.__flush_buffer()

            db = self._coll.database

            # See PYTHON-417, "Sharded GridFS fails with exception: chunks out
            # of order." Inserts via mongos, even if they use a single
            # connection, can succeed out-of-order due to the writebackListener.
            # We mustn't call "filemd5" until all inserts are complete, which
            # we ensure by calling getLastError (and ignoring the result).
            db.error()

            md5 = db.command(
                "filemd5", self._id, root=self._coll.name,
                read_preference=ReadPreference.PRIMARY)["md5"]

            self._file["md5"] = md5
            self._file["length"] = self._position
            self._file["uploadDate"] = datetime.datetime.utcnow()

            return self._coll.files.insert(self._file,
                                           **self._coll._get_wc_override())
        except DuplicateKeyError:
            self._raise_file_exists(self._id)

def start_harvest(self, page=None):
        if not page:
            page = self.pages_complete + 1
        else:
            self.pages_complete = page - 1
        while self.pages_complete < self.total_pages:
            response = self.client.search_agencies(page=page, function=self.function)
            agencies = []
            for result in response['results']:
                print result['agency_id']
                agency = {'agency_id': result['agency_id'], 'title': result['title']}
                agency['agency_status'] = result['agency_status']
                agency['location'] = result['location']
                agency['start_date'] = result['dates']['start_date']
                agency['end_date'] = result['dates']['end_date']
                for function in result['functions']:
                    if function['identifier'] == self.function:
                        agency['function_start'] = function['start_date']
                        agency['function_end'] = function['end_date']
                agencies.append(agency)
                try:
                    self.agencies.insert(result)
                except DuplicateKeyError:
                    pass
            self.functions.update_one({'function': self.function}, {'$push': {'agencies': {'$each': agencies}}}, upsert=True)
            self.pages_complete += 1
            page += 1
            print '{} pages complete'.format(self.pages_complete)
            time.sleep(1)

def process_item(self, item, spider):
        try:
            self.db[self.collection_name].insert(dict(item))
            return item
        except DuplicateKeyError:
            raise DropItem('Duplicated item')

def push_stocks(self,symbol,name,current_price):#??????????????????????
        try:
            self.db.insert({'_id':symbol,'status':self.OUTSIANDING,'????':name,'????':current_price})#????????ID
            print(symbol,name,'??????')
        except errors.DuplicateKeyError as  e:#?????????????
            print(name,'????????')
            pass