We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
tsctl to support TSV files
`tsctl` currently supports reading CSV files, however, often times its easier/less likely data gets broken by using _tab_ delimited files.

This is trivial to implement, e.g - :
1.  add the option `tsv2ts` to [`shell_manager`](https://github.com/hiddenillusion/timesketch/blob/master/tsctl#L424)
2. in `CreateTimelineFromCsv()`have an option for _delimiter_ and if `tsv2ts` was provided, change the CSV [`reader`](https://github.com/hiddenillusion/timesketch/blob/master/tsctl#L412) delimiter to a tab.

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `timesketch/lib/tasks.py`
Content:
```
1 # Copyright 2015 Google Inc. All rights reserved.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 """Celery task for processing Plaso storage files."""
15 
16 import logging
17 import subprocess
18 import traceback
19 
20 from flask import current_app
21 
22 from timesketch import create_app
23 from timesketch import create_celery_app
24 from timesketch.lib.datastores.elastic import ElasticsearchDataStore
25 from timesketch.lib.utils import read_and_validate_csv
26 from timesketch.lib.utils import read_and_validate_jsonl
27 from timesketch.models import db_session
28 from timesketch.models.sketch import SearchIndex
29 from timesketch.models.sketch import Timeline
30 
31 celery = create_celery_app()
32 flask_app = create_app()
33 
34 
35 def _set_timeline_status(index_name, status, error_msg=None):
36     """Helper function to set status for searchindex and all related timelines.
37 
38     Args:
39         index_name: Name of the datastore index.
40         status: Status to set.
41         error_msg: Error message.
42     """
43     # Run within Flask context so we can make database changes
44     with flask_app.app_context():
45         searchindex = SearchIndex.query.filter_by(index_name=index_name).first()
46         timelines = Timeline.query.filter_by(searchindex=searchindex).all()
47 
48         # Set status
49         searchindex.set_status(status)
50         for timeline in timelines:
51             timeline.set_status(status)
52             db_session.add(timeline)
53 
54         # Update description if there was a failure in ingestion
55         if error_msg and status == u'fail':
56             # TODO: Don't overload the description field.
57             searchindex.description = error_msg
58 
59         # Commit changes to database
60         db_session.add(searchindex)
61         db_session.commit()
62 
63 
64 @celery.task(track_started=True)
65 def run_plaso(source_file_path, timeline_name, index_name, source_type,
66               username=None):
67     """Create a Celery task for processing Plaso storage file.
68 
69     Args:
70         source_file_path: Path to plaso storage file.
71         timeline_name: Name of the Timesketch timeline.
72         index_name: Name of the datastore index.
73         source_type: Type of file, csv or jsonl.
74         username: Username of the user who will own the timeline.
75 
76     Returns:
77         String with summary of processed events.
78     """
79     # Log information to Celery
80     logging.info(u'Source type: %s', source_type)
81 
82     cmd = [
83         u'psort.py', u'-o', u'timesketch', source_file_path, u'--name',
84         timeline_name, u'--status_view', u'none', u'--index', index_name
85     ]
86 
87     if username:
88         cmd.append(u'--username')
89         cmd.append(username)
90 
91     # Run psort.py
92     try:
93         cmd_output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)
94     except subprocess.CalledProcessError as e:
95         # Mark the searchindex and timelines as failed and exit the task
96         _set_timeline_status(index_name, status=u'fail', error_msg=e.output)
97         return e.output
98 
99     # Mark the searchindex and timelines as ready
100     _set_timeline_status(index_name, status=u'ready')
101 
102     return cmd_output
103 
104 
105 @celery.task(track_started=True)
106 def run_csv_jsonl(source_file_path, timeline_name, index_name, source_type,
107                   username=None):
108     """Create a Celery task for processing a CSV or JSONL file.
109 
110     Args:
111         source_file_path: Path to CSV or JSONL file.
112         timeline_name: Name of the Timesketch timeline.
113         index_name: Name of the datastore index.
114         source_type: Type of file, csv or jsonl.
115         username: Username of the user who will own the timeline.
116 
117     Returns:
118         Dictionary with count of processed events.
119     """
120     event_type = u'generic_event'  # Document type for Elasticsearch
121     validators = {
122         u'csv': read_and_validate_csv,
123         u'jsonl': read_and_validate_jsonl
124     }
125     read_and_validate = validators.get(source_type)
126 
127     # Log information to Celery
128     logging.info(u'Index name: %s', index_name)
129     logging.info(u'Timeline name: %s', timeline_name)
130     logging.info(u'Source type: %s', source_type)
131     logging.info(u'Document type: %s', event_type)
132     logging.info(u'Owner: %s', username)
133 
134     es = ElasticsearchDataStore(
135         host=current_app.config[u'ELASTIC_HOST'],
136         port=current_app.config[u'ELASTIC_PORT'])
137 
138     # Reason for the broad exception catch is that we want to capture
139     # all possible errors and exit the task.
140     try:
141         es.create_index(index_name=index_name, doc_type=event_type)
142         for event in read_and_validate(source_file_path):
143             es.import_event(index_name, event_type, event)
144         # Import the remaining events
145         total_events = es.import_event(index_name, event_type)
146     except Exception as e:
147         # Mark the searchindex and timelines as failed and exit the task
148         error_msg = traceback.format_exc(e)
149         _set_timeline_status(index_name, status=u'fail', error_msg=error_msg)
150         logging.error(error_msg)
151         return
152 
153     # Set status to ready when done
154     _set_timeline_status(index_name, status=u'ready')
155 
156     return {u'Events processed': total_events}
157
```

Path: `timesketch/lib/utils.py`
Content:
```
1 # Copyright 2015 Google Inc. All rights reserved.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #     http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 """Common functions and utilities."""
15 
16 import colorsys
17 import csv
18 import datetime
19 import json
20 import random
21 import time
22 
23 from dateutil import parser
24 
25 
26 def random_color():
27     """Generates a random color.
28 
29     Returns:
30         Color as string in HEX
31     """
32     hue = random.random()
33     golden_ratio_conjugate = (1 + 5**0.5) / 2
34     hue += golden_ratio_conjugate
35     hue %= 1
36     rgb = tuple(int(i * 256) for i in colorsys.hsv_to_rgb(hue, 0.5, 0.95))
37     return u'{0:02X}{1:02X}{2:02X}'.format(rgb[0], rgb[1], rgb[2])
38 
39 
40 def read_and_validate_csv(path):
41     """Generator for reading a CSV file.
42 
43     Args:
44         path: Path to the CSV file
45     """
46     # Columns that must be present in the CSV file
47     mandatory_fields = [u'message', u'datetime', u'timestamp_desc']
48 
49     with open(path, 'rb') as fh:
50 
51         reader = csv.DictReader(fh)
52         csv_header = reader.fieldnames
53         missing_fields = []
54         # Validate the CSV header
55         for field in mandatory_fields:
56             if field not in csv_header:
57                 missing_fields.append(field)
58         if missing_fields:
59             raise RuntimeError(
60                 u'Missing fields in CSV header: {0:s}'.format(missing_fields))
61         for row in reader:
62             if u'timestamp' not in csv_header and u'datetime' in csv_header:
63                 try:
64                     parsed_datetime = parser.parse(row[u'datetime'])
65                     row[u'timestamp'] = str(
66                         int(time.mktime(parsed_datetime.timetuple())))
67                 except ValueError:
68                     continue
69 
70             yield row
71 
72 
73 def read_and_validate_jsonl(path):
74     """Generator for reading a JSONL (json lines) file.
75 
76     Args:
77         path: Path to the JSONL file
78     """
79     # Fields that must be present in each entry of the JSONL file.
80     mandatory_fields = [u'message', u'datetime', u'timestamp_desc']
81     with open(path, 'rb') as fh:
82 
83         lineno = 0
84         for line in fh:
85             lineno += 1
86             try:
87                 linedict = json.loads(line)
88                 ld_keys = linedict.keys()
89                 if u'datetime' not in ld_keys and u'timestamp' in ld_keys:
90                     epoch = int(str(linedict[u'timestamp'])[:10])
91                     dt = datetime.datetime.fromtimestamp(epoch)
92                     linedict[u'datetime'] = dt.isoformat()
93                 if u'timestamp' not in ld_keys and u'datetime' in ld_keys:
94                     linedict[u'timestamp'] = parser.parse(linedict[u'datetime'])
95 
96                 missing_fields = []
97                 for field in mandatory_fields:
98                     if field not in linedict.keys():
99                         missing_fields.append(field)
100                 if missing_fields:
101                     raise RuntimeError(
102                         u"Missing field(s) at line {0:n}: {1:s}"
103                         .format(lineno, missing_fields))
104 
105                 yield linedict
106 
107             except ValueError as e:
108                 raise RuntimeError(
109                     u"Error parsing JSON at line {0:n}: {1:s}"
110                     .format(lineno, e))
111 
112 
113 def get_validated_indices(indices, sketch_indices):
114     """Exclude any deleted search index references.
115 
116     Args:
117         indices: List of indices from the user
118         sketch_indices: List of indices in the sketch
119 
120     Returns:
121         Set of indices with those removed that is not in the sketch
122     """
123     exclude = set(indices) - set(sketch_indices)
124     if exclude:
125         indices = [index for index in indices if index not in exclude]
126     return indices
127
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/timesketch/lib/tasks.py b/timesketch/lib/tasks.py
--- a/timesketch/lib/tasks.py
+++ b/timesketch/lib/tasks.py
@@ -104,7 +104,7 @@
 
 @celery.task(track_started=True)
 def run_csv_jsonl(source_file_path, timeline_name, index_name, source_type,
-                  username=None):
+                  delimiter=None, username=None):
     """Create a Celery task for processing a CSV or JSONL file.
 
     Args:
@@ -112,6 +112,7 @@
         timeline_name: Name of the Timesketch timeline.
         index_name: Name of the datastore index.
         source_type: Type of file, csv or jsonl.
+        delimiter: Character used as a field separator
         username: Username of the user who will own the timeline.
 
     Returns:
@@ -139,7 +140,7 @@
     # all possible errors and exit the task.
     try:
         es.create_index(index_name=index_name, doc_type=event_type)
-        for event in read_and_validate(source_file_path):
+        for event in read_and_validate(source_file_path, delimiter):
             es.import_event(index_name, event_type, event)
         # Import the remaining events
         total_events = es.import_event(index_name, event_type)
diff --git a/timesketch/lib/utils.py b/timesketch/lib/utils.py
--- a/timesketch/lib/utils.py
+++ b/timesketch/lib/utils.py
@@ -37,18 +37,19 @@
     return u'{0:02X}{1:02X}{2:02X}'.format(rgb[0], rgb[1], rgb[2])
 
 
-def read_and_validate_csv(path):
-    """Generator for reading a CSV file.
+def read_and_validate_csv(path, delimiter):
+    """Generator for reading a CSV or TSV file.
 
     Args:
-        path: Path to the CSV file
+        path: Path to the file
+        delimiter: character used as a field separator
     """
     # Columns that must be present in the CSV file
     mandatory_fields = [u'message', u'datetime', u'timestamp_desc']
 
     with open(path, 'rb') as fh:
 
-        reader = csv.DictReader(fh)
+        reader = csv.DictReader(fh, delimiter=delimiter.decode('string_escape'))
         csv_header = reader.fieldnames
         missing_fields = []
         # Validate the CSV header
@@ -70,7 +71,7 @@
             yield row
 
 
-def read_and_validate_jsonl(path):
+def read_and_validate_jsonl(path, _):
     """Generator for reading a JSONL (json lines) file.
 
     Args:

{"golden_diff": "diff --git a/timesketch/lib/tasks.py b/timesketch/lib/tasks.py\n--- a/timesketch/lib/tasks.py\n+++ b/timesketch/lib/tasks.py\n@@ -104,7 +104,7 @@\n \n @celery.task(track_started=True)\n def run_csv_jsonl(source_file_path, timeline_name, index_name, source_type,\n-                  username=None):\n+                  delimiter=None, username=None):\n     \"\"\"Create a Celery task for processing a CSV or JSONL file.\n \n     Args:\n@@ -112,6 +112,7 @@\n         timeline_name: Name of the Timesketch timeline.\n         index_name: Name of the datastore index.\n         source_type: Type of file, csv or jsonl.\n+        delimiter: Character used as a field separator\n         username: Username of the user who will own the timeline.\n \n     Returns:\n@@ -139,7 +140,7 @@\n     # all possible errors and exit the task.\n     try:\n         es.create_index(index_name=index_name, doc_type=event_type)\n-        for event in read_and_validate(source_file_path):\n+        for event in read_and_validate(source_file_path, delimiter):\n             es.import_event(index_name, event_type, event)\n         # Import the remaining events\n         total_events = es.import_event(index_name, event_type)\ndiff --git a/timesketch/lib/utils.py b/timesketch/lib/utils.py\n--- a/timesketch/lib/utils.py\n+++ b/timesketch/lib/utils.py\n@@ -37,18 +37,19 @@\n     return u'{0:02X}{1:02X}{2:02X}'.format(rgb[0], rgb[1], rgb[2])\n \n \n-def read_and_validate_csv(path):\n-    \"\"\"Generator for reading a CSV file.\n+def read_and_validate_csv(path, delimiter):\n+    \"\"\"Generator for reading a CSV or TSV file.\n \n     Args:\n-        path: Path to the CSV file\n+        path: Path to the file\n+        delimiter: character used as a field separator\n     \"\"\"\n     # Columns that must be present in the CSV file\n     mandatory_fields = [u'message', u'datetime', u'timestamp_desc']\n \n     with open(path, 'rb') as fh:\n \n-        reader = csv.DictReader(fh)\n+        reader = csv.DictReader(fh, delimiter=delimiter.decode('string_escape'))\n         csv_header = reader.fieldnames\n         missing_fields = []\n         # Validate the CSV header\n@@ -70,7 +71,7 @@\n             yield row\n \n \n-def read_and_validate_jsonl(path):\n+def read_and_validate_jsonl(path, _):\n     \"\"\"Generator for reading a JSONL (json lines) file.\n \n     Args:\n", "issue": "tsctl to support TSV files\n`tsctl` currently supports reading CSV files, however, often times its easier/less likely data gets broken by using _tab_ delimited files.\r\n\r\nThis is trivial to implement, e.g - :\r\n1.  add the option `tsv2ts` to [`shell_manager`](https://github.com/hiddenillusion/timesketch/blob/master/tsctl#L424)\r\n2. in `CreateTimelineFromCsv()`have an option for _delimiter_ and if `tsv2ts` was provided, change the CSV [`reader`](https://github.com/hiddenillusion/timesketch/blob/master/tsctl#L412) delimiter to a tab.\n", "before_files": [{"content": "# Copyright 2015 Google Inc. All rights reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"Celery task for processing Plaso storage files.\"\"\"\n\nimport logging\nimport subprocess\nimport traceback\n\nfrom flask import current_app\n\nfrom timesketch import create_app\nfrom timesketch import create_celery_app\nfrom timesketch.lib.datastores.elastic import ElasticsearchDataStore\nfrom timesketch.lib.utils import read_and_validate_csv\nfrom timesketch.lib.utils import read_and_validate_jsonl\nfrom timesketch.models import db_session\nfrom timesketch.models.sketch import SearchIndex\nfrom timesketch.models.sketch import Timeline\n\ncelery = create_celery_app()\nflask_app = create_app()\n\n\ndef _set_timeline_status(index_name, status, error_msg=None):\n    \"\"\"Helper function to set status for searchindex and all related timelines.\n\n    Args:\n        index_name: Name of the datastore index.\n        status: Status to set.\n        error_msg: Error message.\n    \"\"\"\n    # Run within Flask context so we can make database changes\n    with flask_app.app_context():\n        searchindex = SearchIndex.query.filter_by(index_name=index_name).first()\n        timelines = Timeline.query.filter_by(searchindex=searchindex).all()\n\n        # Set status\n        searchindex.set_status(status)\n        for timeline in timelines:\n            timeline.set_status(status)\n            db_session.add(timeline)\n\n        # Update description if there was a failure in ingestion\n        if error_msg and status == u'fail':\n            # TODO: Don't overload the description field.\n            searchindex.description = error_msg\n\n        # Commit changes to database\n        db_session.add(searchindex)\n        db_session.commit()\n\n\n@celery.task(track_started=True)\ndef run_plaso(source_file_path, timeline_name, index_name, source_type,\n              username=None):\n    \"\"\"Create a Celery task for processing Plaso storage file.\n\n    Args:\n        source_file_path: Path to plaso storage file.\n        timeline_name: Name of the Timesketch timeline.\n        index_name: Name of the datastore index.\n        source_type: Type of file, csv or jsonl.\n        username: Username of the user who will own the timeline.\n\n    Returns:\n        String with summary of processed events.\n    \"\"\"\n    # Log information to Celery\n    logging.info(u'Source type: %s', source_type)\n\n    cmd = [\n        u'psort.py', u'-o', u'timesketch', source_file_path, u'--name',\n        timeline_name, u'--status_view', u'none', u'--index', index_name\n    ]\n\n    if username:\n        cmd.append(u'--username')\n        cmd.append(username)\n\n    # Run psort.py\n    try:\n        cmd_output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)\n    except subprocess.CalledProcessError as e:\n        # Mark the searchindex and timelines as failed and exit the task\n        _set_timeline_status(index_name, status=u'fail', error_msg=e.output)\n        return e.output\n\n    # Mark the searchindex and timelines as ready\n    _set_timeline_status(index_name, status=u'ready')\n\n    return cmd_output\n\n\n@celery.task(track_started=True)\ndef run_csv_jsonl(source_file_path, timeline_name, index_name, source_type,\n                  username=None):\n    \"\"\"Create a Celery task for processing a CSV or JSONL file.\n\n    Args:\n        source_file_path: Path to CSV or JSONL file.\n        timeline_name: Name of the Timesketch timeline.\n        index_name: Name of the datastore index.\n        source_type: Type of file, csv or jsonl.\n        username: Username of the user who will own the timeline.\n\n    Returns:\n        Dictionary with count of processed events.\n    \"\"\"\n    event_type = u'generic_event'  # Document type for Elasticsearch\n    validators = {\n        u'csv': read_and_validate_csv,\n        u'jsonl': read_and_validate_jsonl\n    }\n    read_and_validate = validators.get(source_type)\n\n    # Log information to Celery\n    logging.info(u'Index name: %s', index_name)\n    logging.info(u'Timeline name: %s', timeline_name)\n    logging.info(u'Source type: %s', source_type)\n    logging.info(u'Document type: %s', event_type)\n    logging.info(u'Owner: %s', username)\n\n    es = ElasticsearchDataStore(\n        host=current_app.config[u'ELASTIC_HOST'],\n        port=current_app.config[u'ELASTIC_PORT'])\n\n    # Reason for the broad exception catch is that we want to capture\n    # all possible errors and exit the task.\n    try:\n        es.create_index(index_name=index_name, doc_type=event_type)\n        for event in read_and_validate(source_file_path):\n            es.import_event(index_name, event_type, event)\n        # Import the remaining events\n        total_events = es.import_event(index_name, event_type)\n    except Exception as e:\n        # Mark the searchindex and timelines as failed and exit the task\n        error_msg = traceback.format_exc(e)\n        _set_timeline_status(index_name, status=u'fail', error_msg=error_msg)\n        logging.error(error_msg)\n        return\n\n    # Set status to ready when done\n    _set_timeline_status(index_name, status=u'ready')\n\n    return {u'Events processed': total_events}\n", "path": "timesketch/lib/tasks.py"}, {"content": "# Copyright 2015 Google Inc. All rights reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"Common functions and utilities.\"\"\"\n\nimport colorsys\nimport csv\nimport datetime\nimport json\nimport random\nimport time\n\nfrom dateutil import parser\n\n\ndef random_color():\n    \"\"\"Generates a random color.\n\n    Returns:\n        Color as string in HEX\n    \"\"\"\n    hue = random.random()\n    golden_ratio_conjugate = (1 + 5**0.5) / 2\n    hue += golden_ratio_conjugate\n    hue %= 1\n    rgb = tuple(int(i * 256) for i in colorsys.hsv_to_rgb(hue, 0.5, 0.95))\n    return u'{0:02X}{1:02X}{2:02X}'.format(rgb[0], rgb[1], rgb[2])\n\n\ndef read_and_validate_csv(path):\n    \"\"\"Generator for reading a CSV file.\n\n    Args:\n        path: Path to the CSV file\n    \"\"\"\n    # Columns that must be present in the CSV file\n    mandatory_fields = [u'message', u'datetime', u'timestamp_desc']\n\n    with open(path, 'rb') as fh:\n\n        reader = csv.DictReader(fh)\n        csv_header = reader.fieldnames\n        missing_fields = []\n        # Validate the CSV header\n        for field in mandatory_fields:\n            if field not in csv_header:\n                missing_fields.append(field)\n        if missing_fields:\n            raise RuntimeError(\n                u'Missing fields in CSV header: {0:s}'.format(missing_fields))\n        for row in reader:\n            if u'timestamp' not in csv_header and u'datetime' in csv_header:\n                try:\n                    parsed_datetime = parser.parse(row[u'datetime'])\n                    row[u'timestamp'] = str(\n                        int(time.mktime(parsed_datetime.timetuple())))\n                except ValueError:\n                    continue\n\n            yield row\n\n\ndef read_and_validate_jsonl(path):\n    \"\"\"Generator for reading a JSONL (json lines) file.\n\n    Args:\n        path: Path to the JSONL file\n    \"\"\"\n    # Fields that must be present in each entry of the JSONL file.\n    mandatory_fields = [u'message', u'datetime', u'timestamp_desc']\n    with open(path, 'rb') as fh:\n\n        lineno = 0\n        for line in fh:\n            lineno += 1\n            try:\n                linedict = json.loads(line)\n                ld_keys = linedict.keys()\n                if u'datetime' not in ld_keys and u'timestamp' in ld_keys:\n                    epoch = int(str(linedict[u'timestamp'])[:10])\n                    dt = datetime.datetime.fromtimestamp(epoch)\n                    linedict[u'datetime'] = dt.isoformat()\n                if u'timestamp' not in ld_keys and u'datetime' in ld_keys:\n                    linedict[u'timestamp'] = parser.parse(linedict[u'datetime'])\n\n                missing_fields = []\n                for field in mandatory_fields:\n                    if field not in linedict.keys():\n                        missing_fields.append(field)\n                if missing_fields:\n                    raise RuntimeError(\n                        u\"Missing field(s) at line {0:n}: {1:s}\"\n                        .format(lineno, missing_fields))\n\n                yield linedict\n\n            except ValueError as e:\n                raise RuntimeError(\n                    u\"Error parsing JSON at line {0:n}: {1:s}\"\n                    .format(lineno, e))\n\n\ndef get_validated_indices(indices, sketch_indices):\n    \"\"\"Exclude any deleted search index references.\n\n    Args:\n        indices: List of indices from the user\n        sketch_indices: List of indices in the sketch\n\n    Returns:\n        Set of indices with those removed that is not in the sketch\n    \"\"\"\n    exclude = set(indices) - set(sketch_indices)\n    if exclude:\n        indices = [index for index in indices if index not in exclude]\n    return indices\n", "path": "timesketch/lib/utils.py"}], "after_files": [{"content": "# Copyright 2015 Google Inc. All rights reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"Celery task for processing Plaso storage files.\"\"\"\n\nimport logging\nimport subprocess\nimport traceback\n\nfrom flask import current_app\n\nfrom timesketch import create_app\nfrom timesketch import create_celery_app\nfrom timesketch.lib.datastores.elastic import ElasticsearchDataStore\nfrom timesketch.lib.utils import read_and_validate_csv\nfrom timesketch.lib.utils import read_and_validate_jsonl\nfrom timesketch.models import db_session\nfrom timesketch.models.sketch import SearchIndex\nfrom timesketch.models.sketch import Timeline\n\ncelery = create_celery_app()\nflask_app = create_app()\n\n\ndef _set_timeline_status(index_name, status, error_msg=None):\n    \"\"\"Helper function to set status for searchindex and all related timelines.\n\n    Args:\n        index_name: Name of the datastore index.\n        status: Status to set.\n        error_msg: Error message.\n    \"\"\"\n    # Run within Flask context so we can make database changes\n    with flask_app.app_context():\n        searchindex = SearchIndex.query.filter_by(index_name=index_name).first()\n        timelines = Timeline.query.filter_by(searchindex=searchindex).all()\n\n        # Set status\n        searchindex.set_status(status)\n        for timeline in timelines:\n            timeline.set_status(status)\n            db_session.add(timeline)\n\n        # Update description if there was a failure in ingestion\n        if error_msg and status == u'fail':\n            # TODO: Don't overload the description field.\n            searchindex.description = error_msg\n\n        # Commit changes to database\n        db_session.add(searchindex)\n        db_session.commit()\n\n\n@celery.task(track_started=True)\ndef run_plaso(source_file_path, timeline_name, index_name, source_type,\n              username=None):\n    \"\"\"Create a Celery task for processing Plaso storage file.\n\n    Args:\n        source_file_path: Path to plaso storage file.\n        timeline_name: Name of the Timesketch timeline.\n        index_name: Name of the datastore index.\n        source_type: Type of file, csv or jsonl.\n        username: Username of the user who will own the timeline.\n\n    Returns:\n        String with summary of processed events.\n    \"\"\"\n    # Log information to Celery\n    logging.info(u'Source type: %s', source_type)\n\n    cmd = [\n        u'psort.py', u'-o', u'timesketch', source_file_path, u'--name',\n        timeline_name, u'--status_view', u'none', u'--index', index_name\n    ]\n\n    if username:\n        cmd.append(u'--username')\n        cmd.append(username)\n\n    # Run psort.py\n    try:\n        cmd_output = subprocess.check_output(cmd, stderr=subprocess.STDOUT)\n    except subprocess.CalledProcessError as e:\n        # Mark the searchindex and timelines as failed and exit the task\n        _set_timeline_status(index_name, status=u'fail', error_msg=e.output)\n        return e.output\n\n    # Mark the searchindex and timelines as ready\n    _set_timeline_status(index_name, status=u'ready')\n\n    return cmd_output\n\n\n@celery.task(track_started=True)\ndef run_csv_jsonl(source_file_path, timeline_name, index_name, source_type,\n                  delimiter=None, username=None):\n    \"\"\"Create a Celery task for processing a CSV or JSONL file.\n\n    Args:\n        source_file_path: Path to CSV or JSONL file.\n        timeline_name: Name of the Timesketch timeline.\n        index_name: Name of the datastore index.\n        source_type: Type of file, csv or jsonl.\n        delimiter: Character used as a field separator\n        username: Username of the user who will own the timeline.\n\n    Returns:\n        Dictionary with count of processed events.\n    \"\"\"\n    event_type = u'generic_event'  # Document type for Elasticsearch\n    validators = {\n        u'csv': read_and_validate_csv,\n        u'jsonl': read_and_validate_jsonl\n    }\n    read_and_validate = validators.get(source_type)\n\n    # Log information to Celery\n    logging.info(u'Index name: %s', index_name)\n    logging.info(u'Timeline name: %s', timeline_name)\n    logging.info(u'Source type: %s', source_type)\n    logging.info(u'Document type: %s', event_type)\n    logging.info(u'Owner: %s', username)\n\n    es = ElasticsearchDataStore(\n        host=current_app.config[u'ELASTIC_HOST'],\n        port=current_app.config[u'ELASTIC_PORT'])\n\n    # Reason for the broad exception catch is that we want to capture\n    # all possible errors and exit the task.\n    try:\n        es.create_index(index_name=index_name, doc_type=event_type)\n        for event in read_and_validate(source_file_path, delimiter):\n            es.import_event(index_name, event_type, event)\n        # Import the remaining events\n        total_events = es.import_event(index_name, event_type)\n    except Exception as e:\n        # Mark the searchindex and timelines as failed and exit the task\n        error_msg = traceback.format_exc(e)\n        _set_timeline_status(index_name, status=u'fail', error_msg=error_msg)\n        logging.error(error_msg)\n        return\n\n    # Set status to ready when done\n    _set_timeline_status(index_name, status=u'ready')\n\n    return {u'Events processed': total_events}\n", "path": "timesketch/lib/tasks.py"}, {"content": "# Copyright 2015 Google Inc. All rights reserved.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#     http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"Common functions and utilities.\"\"\"\n\nimport colorsys\nimport csv\nimport datetime\nimport json\nimport random\nimport time\n\nfrom dateutil import parser\n\n\ndef random_color():\n    \"\"\"Generates a random color.\n\n    Returns:\n        Color as string in HEX\n    \"\"\"\n    hue = random.random()\n    golden_ratio_conjugate = (1 + 5**0.5) / 2\n    hue += golden_ratio_conjugate\n    hue %= 1\n    rgb = tuple(int(i * 256) for i in colorsys.hsv_to_rgb(hue, 0.5, 0.95))\n    return u'{0:02X}{1:02X}{2:02X}'.format(rgb[0], rgb[1], rgb[2])\n\n\ndef read_and_validate_csv(path, delimiter):\n    \"\"\"Generator for reading a CSV or TSV file.\n\n    Args:\n        path: Path to the file\n        delimiter: character used as a field separator\n    \"\"\"\n    # Columns that must be present in the CSV file\n    mandatory_fields = [u'message', u'datetime', u'timestamp_desc']\n\n    with open(path, 'rb') as fh:\n\n        reader = csv.DictReader(fh, delimiter=delimiter.decode('string_escape'))\n        csv_header = reader.fieldnames\n        missing_fields = []\n        # Validate the CSV header\n        for field in mandatory_fields:\n            if field not in csv_header:\n                missing_fields.append(field)\n        if missing_fields:\n            raise RuntimeError(\n                u'Missing fields in CSV header: {0:s}'.format(missing_fields))\n        for row in reader:\n            if u'timestamp' not in csv_header and u'datetime' in csv_header:\n                try:\n                    parsed_datetime = parser.parse(row[u'datetime'])\n                    row[u'timestamp'] = str(\n                        int(time.mktime(parsed_datetime.timetuple())))\n                except ValueError:\n                    continue\n\n            yield row\n\n\ndef read_and_validate_jsonl(path, _):\n    \"\"\"Generator for reading a JSONL (json lines) file.\n\n    Args:\n        path: Path to the JSONL file\n    \"\"\"\n    # Fields that must be present in each entry of the JSONL file.\n    mandatory_fields = [u'message', u'datetime', u'timestamp_desc']\n    with open(path, 'rb') as fh:\n\n        lineno = 0\n        for line in fh:\n            lineno += 1\n            try:\n                linedict = json.loads(line)\n                ld_keys = linedict.keys()\n                if u'datetime' not in ld_keys and u'timestamp' in ld_keys:\n                    epoch = int(str(linedict[u'timestamp'])[:10])\n                    dt = datetime.datetime.fromtimestamp(epoch)\n                    linedict[u'datetime'] = dt.isoformat()\n                if u'timestamp' not in ld_keys and u'datetime' in ld_keys:\n                    linedict[u'timestamp'] = parser.parse(linedict[u'datetime'])\n\n                missing_fields = []\n                for field in mandatory_fields:\n                    if field not in linedict.keys():\n                        missing_fields.append(field)\n                if missing_fields:\n                    raise RuntimeError(\n                        u\"Missing field(s) at line {0:n}: {1:s}\"\n                        .format(lineno, missing_fields))\n\n                yield linedict\n\n            except ValueError as e:\n                raise RuntimeError(\n                    u\"Error parsing JSON at line {0:n}: {1:s}\"\n                    .format(lineno, e))\n\n\ndef get_validated_indices(indices, sketch_indices):\n    \"\"\"Exclude any deleted search index references.\n\n    Args:\n        indices: List of indices from the user\n        sketch_indices: List of indices in the sketch\n\n    Returns:\n        Set of indices with those removed that is not in the sketch\n    \"\"\"\n    exclude = set(indices) - set(sketch_indices)\n    if exclude:\n        indices = [index for index in indices if index not in exclude]\n    return indices\n", "path": "timesketch/lib/utils.py"}]}