我正在尝试记录列表更新进程的输出,当我在本地使用windows路径时,它可以正常工作,但是当我部署到heroku时,它说它无法从/app根目录中找到目录。我已经相应地修改了路径,并首先查找路径和文件,但它仍然抛出“没有这样的文件或目录:/app/logs/[logfile]
我是否需要仅根据系统根更改路径,还是缺少某些内容?
import psycopg2
import json
import time
import os
from os import path
from os.path import join
from urllib.request import urlopen
from datetime import datetime
class Update:
ts = time.time()
st = datetime.fromtimestamp(ts).strftime('%Y-%m-%d_%H%M%S')
folder_path = "/app/logs"
file_ends_with = ".txt"
how_many_days_old_logs_to_remove = 2
now = time.time()
only_files = []
file_full_path = folder_path # OUTPUT LOG FILE FOR LIST
# SRV1 = '192.168.1.177'
SRV2 = '192.168.1.230'
def __init__(self, uri):
self.uri = uri
@property
def get_db(self):
conn = psycopg2.connect(
host=os.environ.get('DATABASE_STR'),
database=os.environ.get('DATABASE_DB'),
user=os.environ.get('DATABASE_USER'),
password=os.environ.get('DATABASE_PW'),
port=os.environ.get('DATABASE_PORT')
)
return conn
@property
def get_json(self):
with urlopen(self.uri) as resp:
src = resp.read()
data = json.loads(src)
return data
@property
def log_file(self):
file_name = f'iLinkUpdate{self.st}.txt'
output_dir = f'{self.folder_path}'
outfil = f'{output_dir}/{file_name}'
try:
outfil = outfil if not path.exists(output_dir + '/' + file_name) else join(output_dir, file_name)
except Exception as e:
print(f'File does not exist or something: {e}')
return outfil
def write_to_log(self, str, outfil=None):
if outfil is None:
outfil = self.log_file
with open(outfil, mode='a') as myfile:
myfile.write(str)
@property
def rem_old_logs(self, days=2):
file_full_path = self.file_full_path
folder_path = self.folder_path
if days:
how_many_days_old_logs_to_remove = days
else:
how_many_days_old_logs_to_remove = self.how_many_days_old_logs_to_remove
file_ends_with = self.file_ends_with
if not os.path.isdir(folder_path):
raise NotADirectoryError('Source directory not found')
else:
#REMOVE OLD LOG FILES (2 days)
for file in os.listdir(folder_path):
file_full_path = os.path.join(folder_path,file)
if os.path.isfile(file_full_path) and file.endswith(file_ends_with):
#Delete files older than x days
if os.stat(file_full_path).st_mtime < self.now - how_many_days_old_logs_to_remove * 86400:
os.remove(file_full_path)
print(f'\n File Removed : {file_full_path}')
@property
def update_ilinks(self):
def sleep_timer(sec=600): # create sleep timer and invoke update_ilinks again once timer is up
print('sleep')
time.sleep(sec)
self.update_ilinks
try:
s = ''
for ilk in self.get_json: # get json data
serial, poll = ilk['SERIAL_NUM'], ilk['LAST_POLL'] if ilk['LAST_POLL'] else '2001-01-01 12:00'
with self.get_db as conn: # connect to PGdb
# update PG db with timestamp
sql = f"update ilink_inventory set last_poll_date = '{poll}' where (serial_num = '{serial}')"
cursor = conn.cursor()
cursor.execute(sql)
conn.commit()
s = f'{serial}\t{poll}\n'
print(s) # print output to console (debug)
self.write_to_log(s, self.log_file) # write to log file
except Exception as e:
print(f'Exception: {e}')
return sleep_timer() # sleep 10 mins and then run again
@property
def run_program(self):
# self.rem_old_logs
return self.update_ilinks
URI = os.environ.get('URI_TO_JSON')
def main():
u = Update(URI) # URI could change but for now it's pointed at onholdwizard
u.run_program
if __name__ == "__main__":
main()
暂无答案!
目前还没有任何答案,快来回答吧!