def search(db_path: PathInfo,
word_query: Optional[Text]='',
date_start: Optional[Text]=None,
date_stop: Optional[Text]=None,
) -> Iterable[NamedTuple]:
""" Returns the search result as a list of NamedTuple of records.
Accepts database file path and optionally, keywords and date range.
Optional:
word_query: if None (default), not included in search filter.
date_start: if None(default), the earliest date is used.
date_stop: if None (default), the present date is used.
"""
if not date_start:
date_start = int(dt.timestamp(dt.strptime('1970-01-02', '%Y-%m-%d')) * 10**6)
else:
date_start = int(dt.timestamp(dt.strptime(date_start, '%Y-%m-%d')) * 10**6)
if not date_stop:
date_stop = int(dt.utcnow().timestamp() * 10 ** 6)
else:
date_stop = int(dt.timestamp(dt.strptime(date_stop, '%Y-%m-%d')) * 10**6)
word_query = helpers.query_sanitizer(word_query, allowed_chars=[' ', '%', '(', ')', '_'])
sql_query, query_bindings = _make_sql_statement(word_query, date_start, date_stop)
search_results = _run_search(db_path, sql_query, query_bindings)
return search_results
python类timestamp()的实例源码
def isloadingstarted(self):
if not self.stopThread:
if dt.timestamp(dt.now()) - os.stat(
self.LOGPATH + max([f for f in os.listdir(self.LOGPATH)])).st_ctime < 10:
try:
log.info("start loadingscreen")
self.launcher.startLS()
except Riot.LoLException:
self.stopThread = True
else:
self.status("loading started")
QtCore.QTimer().singleShot(1000, self.isgamestarted)
else:
QtCore.QTimer().singleShot(1000, self.isloadingstarted)
else:
log.info("Leaguestate closed")
def upload_token(self, bucket, key=None, expires=3600, policys=None):
assert bucket != None and bucket != "", "invalid bucket"
all_policys = policys or {}
if 'scope' not in all_policys:
if key:
all_policys['scope'] = bucket + ':' + key
else:
all_policys['scope'] = bucket
if 'deadline' not in all_policys:
all_policys['deadline'] = int(
datetime.timestamp(datetime.now())) + expires
json_policys = json_encode(all_policys)
b64_encoded_policys = urlsafe_base64_encode(json_policys)
sha1_sign = hmac_sha1(self.__secret_key, b64_encoded_policys)
b64_encoded_sign = urlsafe_base64_encode(sha1_sign)
upload_token = self.__access_key + ':' + \
bytes_decode(b64_encoded_sign) + ':' + \
bytes_decode(b64_encoded_policys)
return upload_token
database_operations.py 文件源码
项目:united-states-of-browsers
作者: kchawla-pi
项目源码
文件源码
阅读 27
收藏 0
点赞 0
评论 0
def make_database_filepaths(output_db: Union[None, Text],
profiles: Optional[Union[Text, Iterable[Text]]]=None
) -> Dict[Text, PathInfo]:
""" Returns dict of path of files.
Accepts output database name and browser profiles.
output_db: Union[filename.fileext, filename, None]. If None, records are not written to database file.
profiles: Union[browser profile, list of browser profiles, None]. If None, includes all profiles.
returns:
{
'source': Databases to read from {profile name: profile paths},
'sink': database to write to,
'hash': filepath to hashes of urls written,
'source_fieldnames': list of fieldnames in source databases
}
"""
source_db_paths, source_fieldnames, source_search_fields = browser_specific_setup.firefox(profiles=profiles)
appdata_path, sink_db_path, url_hash_log_file = paths_setup.setup_output_db_paths(output_db)
file_paths = {'source': source_db_paths,
'source_fieldnames': [*source_fieldnames, 'last_visit_date_readable'],
'search_fieldnames': source_search_fields,
'sink': sink_db_path,
'hash': url_hash_log_file,
'appdata_path': str(appdata_path),
'utc_offset': (dt.timestamp(dt.now()) - dt.timestamp(dt.utcnow()))/3600,
}
return file_paths # collected in a dict, to be written to a JSON file.
def to_hour_timestamp(datetime):
"""Get timestamp (without minutes and second)
:param datetime: datetime object
:return: timestamp
"""
return int(datetime.timestamp() / 3600) * 3600
def get_current_timestamp():
# type: () -> int
"""
Returns the current timestamp, used to set ``timestamp`` for new
:py:class:`ProposedTransaction` objects.
Split out into a separate function so that it can be mocked during
unit tests.
"""
# Python 3.3 introduced a :py:meth:`datetime.timestamp` method, but
# for compatibility with Python 2, we have to do it the old-fashioned
# way.
# http://stackoverflow.com/q/2775864/
return unix_timestamp(datetime.utcnow().timetuple())
def updateMenu(self):
now = int(datetime.timestamp(datetime.utcnow() + timedelta(hours=9)))
# timedelta.total_seconds(timedelta(hours=1)) ? ???? ??? ????
if now - self.lastUpdateTime > 3600:
self.lastUpdateTime = now
dates, menus = getDatesAndMenus()
for index, day in enumerate(self.weekend):
day.update(date=dates[index], menu=menus[index])
def _get_timestamp():
"""
? http://www.toutiao.com/search_content/ ????????????????
??????????????????????????? datetime.today() ??
??????????????????????
"""
row_timestamp = str(datetime.timestamp(datetime.today()))
return row_timestamp.replace('.', '')[:-3]
def _gen_private_url(self, key, host, expires=3600):
assert host != None and host != "", "download host can' be empty"
if not host.startswith("http://"):
host = "http://" + host
download_url = host + '/' + key
token = self._auth.download_token(download_url, expires=expires)
download_url += '?e=' + \
str(int(datetime.timestamp(datetime.now())) + expires)
download_url += "&token=" + token
return download_url
def download_token(self, download_url, expires=3600):
url = download_url
if download_url.find("?") >= 0:
url += "&e=" + \
str(int(datetime.timestamp(datetime.now())) + expires)
else:
url += '?e=' + \
str(int(datetime.timestamp(datetime.now())) + expires)
download_token = self.access_token(url)
return download_token
def write_meta(note_path, bookmark, note_uuid):
timestamp = int(datetime.timestamp(datetime.now()))
note_title = bookmark['title']
qvr_meta = {}
qvr_meta['title'] = note_title
qvr_meta['uuid'] = note_uuid
qvr_meta['created_at'] = int(bookmark['updated_at'].timestamp())
qvr_meta['updated_at'] = timestamp
qvr_meta['tags'] = bookmark['tags'] + ['Pocket']
with open(str(Path(note_path, 'meta.json')), 'w') as f:
f.write(json.dumps(qvr_meta))
def write_meta(note_path, bookmark, note_uuid):
timestamp = int(datetime.timestamp(datetime.now()))
note_title = bookmark['title']
qvr_meta = {}
qvr_meta['title'] = note_title
qvr_meta['uuid'] = note_uuid
qvr_meta['created_at'] = int(bookmark['updated_at'].timestamp())
qvr_meta['updated_at'] = timestamp
qvr_meta['tags'] = bookmark['tags'] + ['Pocket']
with open(str(Path(note_path, 'meta.json')), 'w') as f:
f.write(json.dumps(qvr_meta))
def _generate_session_id(self):
# TODO salt
from hashlib import md5
t = datetime.timestamp(datetime.now())
return md5(str(t).encode()).hexdigest()
def timestamp(dt):
"""Return the unix timestamp (seconds past the epoch) for datetime dt"""
# This is the equivalent of Python 3.3's datetime.timestamp
try:
return dt.timestamp()
except AttributeError:
if dt.tzinfo is None:
return mktime(dt.timetuple())
else:
return (dt - EPOCH).total_seconds()
def rfc2822date(dt):
"""Turn a datetime into a date string as specified in RFC 2822."""
# This is almost the equivalent of Python 3.3's email.utils.format_datetime,
# but treats naive datetimes as local rather than "UTC with no information ..."
timeval = timestamp(dt)
return formatdate(timeval, usegmt=True)