filenotfounderror

ao218c7q  于 2021-08-20  发布在  Java
关注(0)|答案(0)|浏览(238)

我正在尝试记录列表更新进程的输出,当我在本地使用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()

暂无答案!

目前还没有任何答案,快来回答吧!

相关问题