我有一个 django 应用程序,其中 apschaduler 定期运行脚本,从多封电子邮件接收数据并将其插入到 sqlite3 数据库中。当我在本地主机上运行该项目时,一切正常,但是当我在服务器上运行它时,只有 update_db_html 函数起作用。调度程序运行 update_db_sbd 和 update_db_gonets 脚本,读取邮件,但未更新数据库。可能是什么问题呢?抱歉,如果这是一个愚蠢的问题。
更新数据库的函数
import csv
import requests
from bs4 import BeautifulSoup
from .models import *
from .parse_email import *
def update_db_sbd():
imap_data = iridium_sbd_email("imap.mail.ru", "[email protected]", "...")
msg_list_bin = imap_data[0]
IMEI_list = imap_data[1]
data_list = zip(msg_list_bin, IMEI_list)
for data in data_list:
if data[1] in ('300234069704440', '300234069704480', '300234069705500'):
data_format = 'IceBTC'
Table = TableIceBTC
elif data[1] == '300234069701970':
data_format = 'IceBTC11'
Table = TableIceBTC11
elif data[1] == '300234069704960':
data_format = 'IceST'
Table = TableIceST
else:
data_format = 'permafrost-32T'
Table = TablePermafrost
table_data = {}
with open(f"tables/formats/{data_format}.csv") as f:
reader = csv.reader(f)
for row in reader:
if 'GNSS' in row[0]:
row[0] = row[0][4:]
table_data[row[0]] = parse2float(data[0], int(row[1]), int(row[2]), float(row[3]), float(row[4]))
try:
table_data['mdatetime'] = datetime.strptime(f'{int(table_data["Year"])} {int(table_data["Month"])} '
f'{int(table_data["Day"])} {int(table_data["Hour"])} '
f'{int(table_data["Minute"])} {int(table_data["Second"])}',
'%Y %m %d %H %M %S')
except ValueError:
table_data['mdatetime'] = None
table_data = {key: table_data[key] for key in table_data if key not in ("Year", "Month", "Day", "Hour",
"Minute", "Second")}
table_data['IMEI'] = data[1]
Table.objects.create(**table_data)
def update_db_gonets():
# imap_data = gonets_email("imap.mail.ru", "[email protected]", "...")
imap_data = gonets_email("imap.yandex.ru", "[email protected]", "pcbhacarjecuqmgd")
with open(f"tables/formats/ws_gonets.csv") as f:
reader = csv.reader(f)
rl = list(reader)[0][0].split(';')
for data in imap_data:
row = zip(rl, data)
table_data = {el[0]: el[1] for el in row}
table_data['mdatetime'] = datetime.strptime(table_data['Date'] + table_data['Time'], '%Y-%m-%d%H:%M:%S')
del table_data['Date']
del table_data['Time']
TableWeatherStation.objects.create(**table_data)
def update_db_html():
TableMarker.objects.all().delete()
request = requests.get('http://iotdata.iotstorage.su/viewer.php?carid=3221')
data = request.text
soup = BeautifulSoup(data, 'lxml')
for el in soup.find_all('code'):
data_row = el.text.split('&')
table_data = {el.split('=')[0]: el.split('=')[1] for el in data_row}
try:
table_data['mdatetime'] = datetime.strptime(table_data['mdate'] + table_data['mtime'], "'%Y-%m-%d''%H:%M:%S'")
except ValueError:
continue
del table_data['mdate']
del table_data['mtime']
table_data['Latitude'] = table_data.pop('lat')
table_data['Longitude'] = table_data.pop('lon')
TableMarker.objects.create(**table_data)
Apscheduler
from apscheduler.schedulers.background import BackgroundScheduler
from .update_db import *
def start():
scheduler = BackgroundScheduler()
scheduler.add_job(update_db_sbd, 'interval', seconds=900)
scheduler.add_job(update_db_gonets, 'interval', seconds=3600)
scheduler.add_job(update_db_html, 'interval', seconds=3600)
scheduler.start()