<commit_before>﻿using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using Amazon;
using Amazon.CloudWatchLogs;
using Amazon.CloudWatchLogs.Model;
using Amazon.Runtime;
using Amazon.Runtime.CredentialManagement;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using Tweetinvi;
using Tweetinvi.Streaming;
using Stream = Tweetinvi.Stream;
using TwitterCredentials = LogGenerator.Model.TwitterCredentials;

namespace LogGenerator {

    //--- Classes ---
    public static class JsonExtensions {
        public static bool IsNullOrEmpty(this JToken token) {
            return (token == null) ||
                   (token.Type == JTokenType.Array && !token.HasValues) ||
                   (token.Type == JTokenType.Object && !token.HasValues) ||
                   (token.Type == JTokenType.String && token.ToString() == String.Empty) ||
                   (token.Type == JTokenType.Null);
        }
    }

    public class TwitterStream {
        
        //--- Constants ---
        const string AWS_REGION = "us-west-2";
        const string LOG_GROUP = "/lambda-sharp/log-parser/dev";
        const string LOG_STREAM = "test-log-stream";
        
        //--- Fields ---
        private readonly IAmazonCloudWatchLogs _client;
        private readonly IFilteredStream _stream;
        
        //--- Properties ---
        string SequenceToken {
            get {
                var response = _client.DescribeLogStreamsAsync(new DescribeLogStreamsRequest {
                    LogGroupName = LOG_GROUP
                }).Result;
                return response.LogStreams.First(x => x.LogStreamName == LOG_STREAM).UploadSequenceToken;
            }
        }

        //--- Constructors ---
        public TwitterStream() {
            var region = RegionEndpoint.GetBySystemName(AWS_REGION);
            var chain = new CredentialProfileStoreChain();
            AWSCredentials awsCredentials;
            if(!chain.TryGetAWSCredentials("lambdasharp", out awsCredentials)) {
                throw new Exception("AWS Credentials not found!");
            }
            _client = new AmazonCloudWatchLogsClient(awsCredentials, region);
            SetTwitterCredentials();
            _stream = Stream.CreateFilteredStream();
        }
        
        //--- Methods ---
        public void Run(int sampleSize = -1, string filterValue = "cats") {
            try {
                StartStreaming(SequenceToken, sampleSize, filterValue);
            } catch(Exception e) {
                Console.WriteLine(e);
            }
        }

        private void StartStreaming(string sequenceToken, int sampleSize, string filterValue) {
            var counter = 0;
            var logEventsBatch = new List<InputLogEvent>();
            _stream.AddTrack(filterValue);
            // stream.AddLocation(new Coordinates(-74, 40), new Coordinates(-73, 41));
            _stream.MatchingTweetReceived += (sender, eventArgs) => {
                var json = eventArgs.Tweet.ToJson();
                Console.WriteLine(json);
                var tweetInfo = JsonConvert.DeserializeObject<JObject>(json);
                logEventsBatch.Add(new InputLogEvent {
                    Message = GetLogText(tweetInfo),
                    Timestamp = DateTime.Now
                });
                if(++counter % 10 == 0) {
                    sequenceToken = DispatchLogEvents(logEventsBatch, sequenceToken);
                    logEventsBatch = new List<InputLogEvent>();
                }
                if(--sampleSize == 0) {
                    sequenceToken = DispatchLogEvents(logEventsBatch, sequenceToken);
                    Console.WriteLine("Stoping TwitterStream");
                    Stop();
                }
            };
            _stream.StartStreamMatchingAllConditions();
        }
        
        public string DispatchLogEvents(List<InputLogEvent> logEvents, string sequenceToken) {
            Console.WriteLine("Dispatching Log Events.");
            var request = new PutLogEventsRequest {
                  LogGroupName = LOG_GROUP,
                  LogStreamName = LOG_STREAM,
                  LogEvents = logEvents,
                  SequenceToken = sequenceToken
            };
            var response = _client.PutLogEventsAsync(request).Result;
            return response.NextSequenceToken;
        }    
        
        public void Stop() {
            _stream.StopStream();
        }

        private static string GetLogText(JObject tweetInfo) {
            Console.WriteLine(tweetInfo["text"]);
            var user = tweetInfo["user"];
            var coordinates = tweetInfo["coordinates"];
            var retweetedStatus = tweetInfo["retweeted_status"];
            var entities = tweetInfo["entities"];
            var hashTags = entities["hashtags"];
            var text = new List<string> {
                $"[USER]: The user name is ({user["name"]}), they have ({user["favourites_count"]}) favorite tweets and " +
                    $"have tweeted ({user["statuses_count"]}) times. " +
                    $"They have ({user["friends_count"]}) friends and follow ({user["followers_count"]}) people!!",
                $"[MESSAGE]: The tweet message is: ({tweetInfo["text"]})"
            };
            if(!retweetedStatus.IsNullOrEmpty()) {
                var retweetCount = retweetedStatus["retweet_count"];
                var favoriteCount = retweetedStatus["favorite_count"];<commit_msg>Creating batches of 50 and adding more information to the logs<commit_after>﻿using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using Amazon;
using Amazon.CloudWatchLogs;
using Amazon.CloudWatchLogs.Model;
using Amazon.Runtime;
using Amazon.Runtime.CredentialManagement;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using Tweetinvi;
using Tweetinvi.Streaming;
using Stream = Tweetinvi.Stream;
using TwitterCredentials = LogGenerator.Model.TwitterCredentials;

namespace LogGenerator {

    //--- Classes ---
    public static class JsonExtensions {
        public static bool IsNullOrEmpty(this JToken token) {
            return (token == null) ||
                   (token.Type == JTokenType.Array && !token.HasValues) ||
                   (token.Type == JTokenType.Object && !token.HasValues) ||
                   (token.Type == JTokenType.String && token.ToString() == String.Empty) ||
                   (token.Type == JTokenType.Null);
        }
    }

    public class TwitterStream {
        
        //--- Constants ---
        const string AWS_REGION = "us-west-2";
        const string LOG_GROUP = "/lambda-sharp/log-parser/dev";
        const string LOG_STREAM = "test-log-stream";
        
        //--- Fields ---
        private readonly IAmazonCloudWatchLogs _client;
        private readonly IFilteredStream _stream;
        
        //--- Properties ---
        string SequenceToken {
            get {
                var response = _client.DescribeLogStreamsAsync(new DescribeLogStreamsRequest {
                    LogGroupName = LOG_GROUP
                }).Result;
                return response.LogStreams.First(x => x.LogStreamName == LOG_STREAM).UploadSequenceToken;
            }
        }

        //--- Constructors ---
        public TwitterStream() {
            var region = RegionEndpoint.GetBySystemName(AWS_REGION);
            var chain = new CredentialProfileStoreChain();
            AWSCredentials awsCredentials;
            if(!chain.TryGetAWSCredentials("lambdasharp", out awsCredentials)) {
                throw new Exception("AWS Credentials not found!");
            }
            _client = new AmazonCloudWatchLogsClient(awsCredentials, region);
            SetTwitterCredentials();
            _stream = Stream.CreateFilteredStream();
        }
        
        //--- Methods ---
        public void Run(int sampleSize = -1, string filterValue = "cats") {
            try {
                StartStreaming(SequenceToken, sampleSize, filterValue);
            } catch(Exception e) {
                Console.WriteLine(e);
            }
        }

        private void StartStreaming(string sequenceToken, int sampleSize, string filterValue) {
            var counter = 0;
            var logEventsBatch = new List<InputLogEvent>();
            _stream.AddTrack(filterValue);
            // stream.AddLocation(new Coordinates(-74, 40), new Coordinates(-73, 41));
            _stream.MatchingTweetReceived += (sender, eventArgs) => {
                var json = eventArgs.Tweet.ToJson();
                Console.WriteLine(json);
                var tweetInfo = JsonConvert.DeserializeObject<JObject>(json);
                logEventsBatch.Add(new InputLogEvent {
                    Message = GetLogText(tweetInfo),
                    Timestamp = DateTime.Now
                });
                if(++counter % 50 == 0) {
                    sequenceToken = DispatchLogEvents(logEventsBatch, sequenceToken);
                    logEventsBatch = new List<InputLogEvent>();
                }
                if(--sampleSize == 0) {
                    sequenceToken = DispatchLogEvents(logEventsBatch, sequenceToken);
                    Console.WriteLine("Stoping TwitterStream");
                    Stop();
                }
            };
            _stream.StartStreamMatchingAllConditions();
        }
        
        public string DispatchLogEvents(List<InputLogEvent> logEvents, string sequenceToken) {
            Console.WriteLine("Dispatching Log Events.");
            var request = new PutLogEventsRequest {
                  LogGroupName = LOG_GROUP,
                  LogStreamName = LOG_STREAM,
                  LogEvents = logEvents,
                  SequenceToken = sequenceToken
            };
            var response = _client.PutLogEventsAsync(request).Result;
            return response.NextSequenceToken;
        }    
        
        public void Stop() {
            _stream.StopStream();
        }

        private static string GetLogText(JObject tweetInfo) {
            Console.WriteLine(tweetInfo["text"]);
            var user = tweetInfo["user"];
            var coordinates = tweetInfo["coordinates"];
            var retweetedStatus = tweetInfo["retweeted_status"];
            var entities = tweetInfo["entities"];
            var hashTags = entities["hashtags"];
            var tweetDate = tweetInfo["created_at"].ToString();
            var userDate = user["created_at"].ToString();
            var text = new List<string> {
                $"[USER]: The user name is ({user["name"]}), they have ({user["favourites_count"]}) favorite tweets and " +
                    $"have tweeted ({user["statuses_count"]}) times. " +
                    $"They have ({user["friends_count"]}) friends and follow ({user["followers_count"]}) people!!" + 
                    $"This user was created on ({DateTime.Parse(userDate):yyyy-MM-dd HH:mm:ss})",
                $"[MESSAGE]: The tweet by ({user["name"]}) is: ({tweetInfo["text"]}) and it was tweeted at ({DateTime.Parse(tweetDate):yyyy-MM-dd HH:mm:ss})"
            };
            if(!retweetedStatus.IsNullOrEmpty()) {
                var retweetCount = retweetedStatus["retweet_count"];
                var favoriteCount = retweetedStatus["favorite_count"];
{'fraction_non_alphanumeric': 0.10443402070979732, 'fraction_numerical': 0.002832109036197894, 'mean_word_length': 1.9183884297520661, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5422675', 'n_tokens_mistral': 2884, 'n_tokens_neox': 2744, 'n_words': 614}